# 用Claude和Python构建技能驱动的金融分析智能体

- 来源：MarkTechPost（RSS）
- 作者：Sana Hassan
- 发布时间：2026-07-28 02:08
- AIHOT 分数：72
- AIHOT 标记：精选
- AIHOT 链接：https://aihot.virxact.com/items/cms3k7su000n8roiytkztsme7
- 原文链接：https://www.marktechpost.com/2026/07/27/designing-skill-driven-financial-analysis-agents-with-claude-python-mcp-connectors-and-automated-deliverables

## 精选理由

这个教程把手教你将 Anthropic 的金融技能库打包成可运行的 Python 代理，不是概念演示而是完整工作流，做金融 AI 落地的可以直接抄。

## AI 摘要

本教程基于Anthropic的financial-services仓库，用纯Python复现其技能驱动架构。通过解析SKILL.md文件构建可搜索技能注册表，并创建可复用SkillAgent，将金融分析剧本注入Anthropic Messages API，支持迭代工具调用循环。

## 正文

在本教程中，我们围绕 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 或托管智能体中的受控生产运行时，来扩展该系统。