สร้าง Agent วิเคราะห์การเงินอัจฉริยะด้วย Claude, Python และ MCP Connectors ตามแนวทาง Anthropic

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)

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], "...")

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

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)

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)

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.
""")