在本教程中，我们围绕 Anthropic 的 financial-services 仓库构建了一个高级工作流，并用纯 Python 复现了其基于技能（skill）的架构。我们首先安装所需的库、克隆仓库，并以编程方式映射其智能体（agents）、垂直插件（vertical plugins）、合作伙伴集成（partner integrations）、托管智能体操作手册（managed-agent cookbooks）以及财务分析技能（financial analysis skills）。接着，我们将仓库中的 SKILL.md 文件解析成一个可搜索的注册表，并构建一个可复用的 SkillAgent，该智能体将选定的财务操作手册注入 Anthropic Messages API，同时支持用于 Python 计算和文件生成的迭代工具调用循环。利用这一架构，我们执行了一次合成现金流折现估值，生成了 WACC 和终值增长敏感性热力图，进行了可比公司分析并输出格式化 Excel 文件，起草了一份私募股权投资委员会备忘录，并检查了一份托管智能体的部署规范，而未发送实际的部署请求。

import subprocess, sys, os, io, re, json, glob, textwrap, contextlib, pathlib
def sh(cmd):
   print(f"$ {cmd}")
   r = subprocess.run(cmd, shell=True, capture_output=True, text=True)
   if r.returncode != 0:
       print(r.stderr[-1500:])
   return r
sh(f"{sys.executable} -m pip install -q anthropic pandas openpyxl pyyaml matplotlib")
import pandas as pd
import yaml
import matplotlib.pyplot as plt
REPO_URL = "https://github.com/anthropics/financial-services.git"
REPO_DIR = "financial-services"
if not os.path.isdir(REPO_DIR):
   sh(f"git clone --depth 1 {REPO_URL} {REPO_DIR}")
else:
   print("Repo already cloned — skipping.")
def get_api_key():
   try:
       from google.colab import userdata
       k = userdata.get("ANTHROPIC_API_KEY")
       if k:
           return k
   except Exception:
       pass
   if os.environ.get("ANTHROPIC_API_KEY"):
       return os.environ["ANTHROPIC_API_KEY"]
   from getpass import getpass
   return getpass("Enter your Anthropic API key: ")
os.environ["ANTHROPIC_API_KEY"] = get_api_key()
import anthropic
client = anthropic.Anthropic()
MODEL = "claude-sonnet-4-6"
print("SDK ready. Model:", MODEL)

我们安装所需的 Python 库，克隆 Anthropic 的 financial-services 仓库，并准备好 Google Colab 运行环境以供执行。我们从 Colab 密钥、环境变量或安全交互式提示中获取 Anthropic API 密钥。然后，我们初始化官方的 Anthropic SDK，并选择驱动财务分析工作流的 Claude 模型。

def repo_map(root=REPO_DIR):
   rows = []
   for kind, pattern in [
       ("agent",   f"{root}/plugins/agent-plugins/*"),
       ("vertical",f"{root}/plugins/vertical-plugins/*"),
       ("partner", f"{root}/plugins/partner-built/*"),
       ("cookbook",f"{root}/managed-agent-cookbooks/*"),
   ]:
       for p in sorted(glob.glob(pattern)):
           if not os.path.isdir(p):
               continue
           skills   = glob.glob(f"{p}/**/SKILL.md", recursive=True)
           commands = glob.glob(f"{p}/commands/*.md")
           rows.append({"type": kind, "name": os.path.basename(p),
                        "skills": len(skills), "commands": len(commands)})
   return pd.DataFrame(rows)
print("\n=== REPO MAP ===")
repo_df = repo_map()
print(repo_df.to_string(index=False))
mcp_files = glob.glob(f"{REPO_DIR}/plugins/**/.mcp.json", recursive=True)
for f in mcp_files[:1]:
   print(f"\n=== MCP CONNECTORS ({f}) ===")
   try:
       cfg = json.load(open(f))
       for name, srv in cfg.get("mcpServers", cfg).items():
           print(f"  {name:<14} -> {srv.get('url', srv)}")
   except Exception as e:
       print("  (could not parse:", e, ")")
FRONTMATTER = re.compile(r"^---\s*\n(.*?)\n---\s*\n", re.S)
class Skill:
   def __init__(self, path):
       self.path = path
       raw = open(path, encoding="utf-8", errors="replace").read()
       m = FRONTMATTER.match(raw)
       meta = {}
       if m:
           try:
               meta = yaml.safe_load(m.group(1)) or {}
           except Exception:
               meta = {}
       self.name = str(meta.get("name") or pathlib.Path(path).parent.name)
       self.description = str(meta.get("description", ""))[:300]
       self.body = raw[m.end():] if m else raw
   def __repr__(self):
       return f"<Skill {self.name}>"
class SkillRegistry:
   def __init__(self, root=REPO_DIR):
       paths  = sorted(glob.glob(f"{root}/plugins/vertical-plugins/**/SKILL.md", recursive=True))
       paths += sorted(glob.glob(f"{root}/plugins/**/SKILL.md", recursive=True))
       self.skills = {}
       for p in paths:
           s = Skill(p)
           self.skills.setdefault(s.name.lower(), s)
   def find(self, query):
       q = query.lower()
       hits = [s for k, s in self.skills.items() if q in k]
       if not hits:
           hits = [s for s in self.skills.values() if q in s.description.lower()]
       return hits
   def get(self, query):
       hits = self.find(query)
       if not hits:
           raise KeyError(f"No skill matching '{query}'. "
                          f"Available: {sorted(self.skills)[:40]}")
       return hits[0]
registry = SkillRegistry()
print(f"\nLoaded {len(registry.skills)} unique skills.")
print("Sample:", sorted(registry.skills)[:12], "...")

我们检查仓库结构，识别其智能体插件（agent plugins）、垂直插件（vertical plugins）、合作伙伴集成（partner integrations）、托管智能体操作手册（managed-agent cookbooks）以及可用命令。我们找到 MCP 配置文件，并展示仓库中定义的外部财务数据连接器。接着，我们解析每个 SKILL.md 文件，提取其 YAML 元数据和方法论，并注册每一个独特的技能，以便进行可搜索的访问。

os.makedirs("outputs", exist_ok=True)
TOOLS = [
   {
       "name": "run_python",
       "description": ("Execute Python code and return stdout. pandas as pd "
                       "and numpy as np are pre-imported. Use print() to "
                       "return results. State persists across calls."),
       "input_schema": {
           "type": "object",
           "properties": {"code": {"type": "string"}},
           "required": ["code"],
       },
   },
   {
       "name": "save_file",
       "description": "Save text content to outputs/<filename>.",
       "input_schema": {
           "type": "object",
           "properties": {"filename": {"type": "string"},
                          "content":  {"type": "string"}},
           "required": ["filename", "content"],
       },
   },
]
_PY_NS = {}
def _tool_run_python(code):
   import numpy as np
   _PY_NS.setdefault("pd", pd); _PY_NS.setdefault("np", np)
   buf = io.StringIO()
   try:
       with contextlib.redirect_stdout(buf):
           exec(code, _PY_NS)
       out = buf.getvalue()
       return out[:6000] if out else "(no stdout — use print())"
   except Exception as e:
       return f"ERROR: {type(e).__name__}: {e}"
def _tool_save_file(filename, content):
   safe = os.path.basename(filename)
   path = os.path.join("outputs", safe)
   open(path, "w", encoding="utf-8").write(content)
   return f"Saved {path} ({len(content)} chars)"
DISPATCH = {"run_python": lambda i: _tool_run_python(i["code"]),
           "save_file":  lambda i: _tool_save_file(i["filename"], i["content"])}
BASE_SYSTEM = """You are a financial analyst assistant operating with the
skill playbooks provided below (from Anthropic's financial-services repo).
Follow the skill's methodology, conventions, and output format closely.
Use the run_python tool for all numerical work — never do arithmetic in
your head. Use save_file for final deliverables. All work is a DRAFT for
human review; do not present it as investment advice."""
class SkillAgent:
   """Minimal reproduction of Cowork's skill-firing: chosen skills are
   concatenated into the system prompt; the agent then runs a standard
   tool-use loop against the Messages API until the model stops."""
   def __init__(self, skill_queries, max_skill_chars=12000, verbose=True):
       self.skills = [registry.get(q) for q in skill_queries]
       blocks = []
       for s in self.skills:
           blocks.append(f"\n\n===== SKILL: {s.name} =====\n"
                         f"{s.description}\n{s.body[:max_skill_chars]}")
       self.system = BASE_SYSTEM + "".join(blocks)
       self.verbose = verbose
   def run(self, prompt, max_turns=12):
       messages = [{"role": "user", "content": prompt}]
       for turn in range(max_turns):
           resp = client.messages.create(
               model=MODEL, max_tokens=8000,
               system=self.system, tools=TOOLS, messages=messages)
           messages.append({"role": "assistant", "content": resp.content})
           if resp.stop_reason != "tool_use":
               final = "".join(b.text for b in resp.content
                               if b.type == "text")
               return final, messages
           results = []
           for block in resp.content:
               if block.type == "tool_use":
                   if self.verbose:
                       print(f"  [turn {turn}] tool: {block.name}")
                   out = DISPATCH[block.name](block.input)
                   results.append({"type": "tool_result",
                                   "tool_use_id": block.id,
                                   "content": str(out)})
           messages.append({"role": "user", "content": results})
       return "(hit max_turns)", messages

我们定义工具，使 Claude 能够执行 Python 计算并将生成的交付物保存到 Colab 环境中。我们构建一个持久的 Python 命名空间，使得数值模型、表格和中间变量在智能体的多次交互回合中保持可用。然后，我们创建 SkillAgent 类，将选定的财务操作手册注入其系统提示词，并管理 Anthropic Messages API 的工具调用循环。

SAMPLE_CO = """
Target: 'Meridian Software' (synthetic). FY2025 actuals, $mm:
Revenue 850 (grew 18% y/y) | EBITDA margin 27% | D&A 4% of rev
CapEx 5% of rev | NWC change 1% of rev growth | Tax rate 24%
Net debt 320 | Diluted shares 92mm
Assumptions: revenue growth fades 18% -> 6% linearly over 5 yrs,
EBITDA margin expands 100bps total, WACC 9.5%, terminal growth 2.5%.
"""
print("\n" + "="*76 + "\nDEMO A — DCF (dcf-model skill)\n" + "="*76)
try:
   dcf_agent = SkillAgent(["dcf"])
   dcf_answer, _ = dcf_agent.run(
       "Run a 5-year unlevered DCF per the skill playbook on this company:\n"
       + SAMPLE_CO +
       "\nCompute enterprise value, equity value, and implied share price "
       "with run_python. Then print a WACC (8.5%-10.5%, 50bp steps) x "
       "terminal growth (1.5%-3.5%, 50bp steps) sensitivity grid of implied "
       "share price as a JSON object under the marker SENS_JSON:, and give "
       "a concise summary.")
   print("\n--- DCF RESULT ---\n", dcf_answer[:3000])
   m = re.search(r"SENS_JSON:\s*(\{.*\})", dcf_answer, re.S)
   if m:
       grid = json.loads(m.group(1))
       sens = pd.DataFrame(grid)
       sens = sens.apply(pd.to_numeric, errors="coerce")
       fig, ax = plt.subplots(figsize=(7, 4))
       im = ax.imshow(sens.values, cmap="RdYlGn", aspect="auto")
       ax.set_xticks(range(len(sens.columns)), sens.columns)
       ax.set_yticks(range(len(sens.index)), sens.index)
       ax.set_xlabel("Terminal growth"); ax.set_ylabel("WACC")
       ax.set_title("Implied share price sensitivity ($)")
       for i in range(sens.shape[0]):
           for j in range(sens.shape[1]):
               v = sens.values[i, j]
               if pd.notna(v):
                   ax.text(j, i, f"{v:,.0f}", ha="center", va="center",
                           fontsize=8)
       fig.colorbar(im); plt.tight_layout(); plt.show()
except Exception as e:
   print("Demo A skipped:", e)

我们提供合成运营假设，并指示 DCF 技能智能体构建一个五年期无杠杆现金流估值模型。我们使用 Python 执行工具计算企业价值、股权价值、隐含股价以及二维敏感性矩阵。随后，我们提取结构化的敏感性结果，并通过热力图可视化 WACC、终值增长率与隐含估值之间的关系。

PEERS = """
Synthetic peer set ($mm except per-share):
Ticker  Price  Shares  NetDebt  Rev_NTM  EBITDA_NTM  EPS_NTM
ALFA    64.2   210     450      2900     820         3.10
BRVO    28.7   540     -120     4100     980         1.45
CHRL    112.5  95      760      1850     610         5.60
DLTA    41.9   330     210      2600     700         2.05
"""
print("\n" + "="*76 + "\nDEMO B — COMPS (comps-analysis skill) -> Excel\n" + "="*76)
try:
   comps_agent = SkillAgent(["comps"])
   comps_answer, _ = comps_agent.run(
       "Per the comps skill, compute EV, EV/Revenue, EV/EBITDA and P/E "
       "(all NTM) for these peers with run_python:\n" + PEERS +
       "\nThen print the full comps table plus min/25th/median/75th/max "
       "summary stats as JSON under the marker COMPS_JSON: with keys "
       "'table' (list of row dicts) and 'stats' (dict of dicts).")
   print("\n--- COMPS NARRATIVE ---\n", comps_answer[:1500])
   m = re.search(r"COMPS_JSON:\s*(\{.*\})", comps_answer, re.S)
   if m:
       payload = json.loads(m.group(1))
       table = pd.DataFrame(payload["table"])
       stats = pd.DataFrame(payload["stats"])
       xlsx = "outputs/comps_analysis.xlsx"
       with pd.ExcelWriter(xlsx, engine="openpyxl") as xl:
           table.to_excel(xl, sheet_name="Comps", index=False)
           stats.to_excel(xl, sheet_name="Summary Stats")
       from openpyxl import load_workbook
       from openpyxl.styles import Font, PatternFill
       wb = load_workbook(xlsx)
       for ws in wb.worksheets:
           for cell in ws[1]:
               cell.font = Font(bold=True, color="FFFFFF")
               cell.fill = PatternFill("solid", start_color="1F4E79")
           for col in ws.columns:
               w = max(len(str(c.value)) for c in col if c.value is not None)
               ws.column_dimensions[col[0].column_letter].width = w + 3
       wb.save(xlsx)
       print("Wrote", xlsx)
       print(table.to_string(index=False))
except Exception as e:
   print("Demo B skipped:", e)

我们提供一组合成可比公司，并计算企业价值、EV/收入、EV/EBITDA 以及市盈率倍数。我们将智能体输出的结构化 JSON 响应转换为详细的可比公司数据框和汇总统计量数据框。接着，我们将分析结果导出为多工作表 Excel 工作簿，并应用专业表头格式与自动列宽调整。

print("\n" + "="*76 + "\nDEMO C — IC MEMO (ic-memo skill)\n" + "="*76)
try:
   ic_agent = SkillAgent(["ic-memo"])
   ic_answer, _ = ic_agent.run(
       "Draft a first-round IC memo per the skill for a hypothetical "
       "buyout of Meridian Software (see facts below). Base the valuation "
       "framing on ~11x EV/EBITDA entry, 45% leverage, 5-yr hold. Use "
       "run_python for any quick math (e.g., rough MOIC/IRR math) and "
       "save the memo with save_file as ic_memo_meridian.md.\n" + SAMPLE_CO)
   print("\n--- IC MEMO (first 1500 chars of reply) ---\n", ic_answer[:1500])
   memo_path = "outputs/ic_memo_meridian.md"
   if os.path.exists(memo_path):
       print(f"\nMemo saved -> {memo_path} "
             f"({os.path.getsize(memo_path)} bytes)")
except Exception as e:
   print("Demo C skipped:", e)
print("\n" + "="*76 + "\nPART 7 — MANAGED AGENT COOKBOOK (dry run)\n" + "="*76)
yamls = sorted(glob.glob(f"{REPO_DIR}/managed-agent-cookbooks/*/agent.yaml")) \
     + sorted(glob.glob(f"{REPO_DIR}/managed-agent-cookbooks/*/*.yaml"))
if yamls:
   path = yamls[0]
   print("Inspecting:", path)
   try:
       spec = yaml.safe_load(open(path))
       print(json.dumps(spec, indent=2, default=str)[:2500])
       print("\nDeploy flow: resolve file refs -> upload skills -> create "
             "leaf-worker subagents -> POST orchestrator to /v1/agents "
             "(see scripts/orchestrate.py for the handoff_request event loop).")
   except Exception as e:
       print("Could not parse yaml:", e)
else:
   print("No cookbook yaml found on this branch — see "
         "managed-agent-cookbooks/ READMEs on GitHub.")
print("\n" + "="*76)
print("DONE. Artifacts in ./outputs/:", os.listdir("outputs"))
print("""
Where to go next:
* Swap SAMPLE_CO / PEERS for real data via the repo's MCP connectors
  (Daloopa, FactSet, S&P, Morningstar, PitchBook...) — subscriptions apply.
* Load other skills: SkillAgent(["lbo"]), ["merger"], ["earnings"],
  ["rebalance"], ["tlh"], ["kyc"] ... — see `sorted(registry.skills)`.
* Stack skills: SkillAgent(["comps", "dcf"]) for a football-field workflow.
* For production, install as a Cowork plugin or deploy via Managed Agents
  instead of this Colab loop — same skills, governed runtime.
All outputs are drafts for qualified human review — not investment advice.
""")

我们将投资委员会备忘录技能应用于一个假设的软件收购案例，并使用 Python 计算支持性回报指标。我们将生成的备忘录保存为 Markdown 交付物，并确认该文件已存在于输出目录中。随后，我们查看一个托管智能体操作手册，展示部署规范，最后回顾生成的工件及可能的生产环境扩展方案。

完成本教程后，我们实现了一个基于 Colab 的实用近似方案，模拟 Anthropic 的金融服务技能与智能体框架，同时保留了该仓库以方法论驱动的金融分析方法。我们将结构化技能发现、动态系统提示词构建、持久化 Python 执行、基于 API 的工具编排以及自动化交付物生成整合到一个可复用的工作流中。我们还展示了同一智能体架构如何支持多种金融用例，包括 DCF 估值、交易可比分析、敏感性测试、Excel 报告以及投资委员会备忘录准备。在此基础上，我们可以通过加载更多技能、组合多种估值手册、通过 MCP 集成连接授权金融数据提供商，以及将教程沙盒替换为 Claude Code、Cowork 或托管智能体中的受控生产运行时，来扩展该系统。

---