Designing Skill-Driven Financial Analysis Agents with Claude, Python, MCP Connectors, and Automated Deliverables

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, textual content=True)
   if r.returncode != 0:
       print(r.stderr[-1500:])
   return r
sh(f"{sys.executable} -m pip set up -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():
   strive:
       from google.colab import userdata
       ok = userdata.get("ANTHROPIC_API_KEY")
       if ok:
           return ok
   besides Exception:
       move
   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
consumer = anthropic.Anthropic()
MODEL = "claude-sonnet-4-6"
print("SDK prepared. Model:", MODEL)

def repo_map(root=REPO_DIR):
   rows = []
   for sort, sample 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(sample)):
           if not os.path.isdir(p):
               proceed
           expertise   = glob.glob(f"{p}/**/SKILL.md", recursive=True)
           instructions = glob.glob(f"{p}/instructions/*.md")
           rows.append({"kind": sort, "title": os.path.basename(p),
                        "expertise": len(expertise), "instructions": len(instructions)})
   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}) ===")
   strive:
       cfg = json.load(open(f))
       for title, srv in cfg.get("mcpServers", cfg).objects():
           print(f"  {title:<14} -> {srv.get('url', srv)}")
   besides Exception as e:
       print("  (couldn't parse:", e, ")")
FRONTMATTER = re.compile(r"^---s*n(.*?)n---s*n", re.S)
class Skill:
   def __init__(self, path):
       self.path = path
       uncooked = open(path, encoding="utf-8", errors="substitute").learn()
       m = FRONTMATTER.match(uncooked)
       meta = {}
       if m:
           strive:
               meta = yaml.safe_load(m.group(1)) or {}
           besides Exception:
               meta = {}
       self.title = str(meta.get("title") or pathlib.Path(path).mother or father.title)
       self.description = str(meta.get("description", ""))[:300]
       self.physique = uncooked[m.end():] if m else uncooked
   def __repr__(self):
       return f"<Skill {self.title}>"
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.expertise = {}
       for p in paths:
           s = Skill(p)
           self.expertise.setdefault(s.title.decrease(), s)
   def discover(self, question):
       q = question.decrease()
       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, question):
       hits = self.discover(question)
       if not hits:
           elevate KeyError(f"No talent matching '{question}'. "
                          f"Available: {sorted(self.expertise)[:40]}")
       return hits[0]
registry = SkillRegistry()
print(f"nLoaded {len(registry.expertise)} distinctive expertise.")
print("Sample:", sorted(registry.expertise)[: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"],
       },
   },
   {
       "title": "save_file",
       "description": "Save textual content content material to outputs/<filename>.",
       "input_schema": {
           "kind": "object",
           "properties": {"filename": {"kind": "string"},
                          "content material":  {"kind": "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()
   strive:
       with contextlib.redirect_stdout(buf):
           exec(code, _PY_NS)
       out = buf.getvalue()
       return out[:6000] if out else "(no stdout — use print())"
   besides Exception as e:
       return f"ERROR: {kind(e).__name__}: {e}"
def _tool_save_file(filename, content material):
   protected = os.path.basename(filename)
   path = os.path.be a part of("outputs", protected)
   open(path, "w", encoding="utf-8").write(content material)
   return f"Saved {path} ({len(content material)} 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 monetary analyst assistant working with the
talent playbooks offered under (from Anthropic's financial-services repo).
Follow the talent's methodology, conventions, and output format intently.
Use the run_python instrument for all numerical work — by no means do arithmetic in
your head. Use save_file for ultimate deliverables. All work is a DRAFT for
human evaluate; don't current it as funding recommendation."""
class SkillAgent:
   """Minimal copy of Cowork's skill-firing: chosen expertise are
   concatenated into the system immediate; the agent then runs a normal
   tool-use loop towards the Messages API till the mannequin stops."""
   def __init__(self, skill_queries, max_skill_chars=12000, verbose=True):
       self.expertise = [registry.get(q) for q in skill_queries]
       blocks = []
       for s in self.expertise:
           blocks.append(f"nn===== SKILL: {s.title} =====n"
                         f"{s.description}n{s.physique[:max_skill_chars]}")
       self.system = BASE_SYSTEM + "".be a part of(blocks)
       self.verbose = verbose
   def run(self, immediate, max_turns=12):
       messages = [{"role": "user", "content": prompt}]
       for flip in vary(max_turns):
           resp = consumer.messages.create(
               mannequin=MODEL, max_tokens=8000,
               system=self.system, instruments=TOOLS, messages=messages)
           messages.append({"position": "assistant", "content material": resp.content material})
           if resp.stop_reason != "tool_use":
               ultimate = "".be a part of(b.textual content for b in resp.content material
                               if b.kind == "textual content")
               return ultimate, messages
           outcomes = []
           for block in resp.content material:
               if block.kind == "tool_use":
                   if self.verbose:
                       print(f"  [turn {turn}] instrument: {block.title}")
                   out = DISPATCH[block.name](block.enter)
                   outcomes.append({"kind": "tool_result",
                                   "tool_use_id": block.id,
                                   "content material": str(out)})
           messages.append({"position": "person", "content material": outcomes})
       return "(hit max_turns)", messages

SAMPLE_CO = """
Target: 'Meridian Software' (artificial). 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 progress | Tax fee 24%
Net debt 320 | Diluted shares 92mm
Assumptions: income progress fades 18% -> 6% linearly over 5 yrs,
EBITDA margin expands 100bps complete, WACC 9.5%, terminal progress 2.5%.
"""
print("n" + "="*76 + "nDEMO A — DCF (dcf-model talent)n" + "="*76)
strive:
   dcf_agent = SkillAgent(["dcf"])
   dcf_answer, _ = dcf_agent.run(
       "Run a 5-year unlevered DCF per the talent playbook on this firm:n"
       + SAMPLE_CO +
       "nCompute enterprise worth, fairness worth, and implied share value "
       "with run_python. Then print a WACC (8.5%-10.5%, 50bp steps) x "
       "terminal progress (1.5%-3.5%, 50bp steps) sensitivity grid of implied "
       "share value as a JSON object below the marker SENS_JSON:, and give "
       "a concise abstract.")
   print("n--- DCF RESULT ---n", dcf_answer[:3000])
   m = re.search(r"SENS_JSON:s*({.*})", dcf_answer, re.S)
   if m:
       grid = json.hundreds(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", side="auto")
       ax.set_xticks(vary(len(sens.columns)), sens.columns)
       ax.set_yticks(vary(len(sens.index)), sens.index)
       ax.set_xlabel("Terminal progress"); ax.set_ylabel("WACC")
       ax.set_title("Implied share value sensitivity ($)")
       for i in vary(sens.form[0]):
           for j in vary(sens.form[1]):
               v = sens.values[i, j]
               if pd.notna(v):
                   ax.textual content(j, i, f"{v:,.0f}", ha="heart", va="heart",
                           fontsize=8)
       fig.colorbar(im); plt.tight_layout(); plt.present()
besides Exception as e:
   print("Demo A skipped:", e)

PEERS = """
Synthetic peer set ($mm besides 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 talent) -> Exceln" + "="*76)
strive:
   comps_agent = SkillAgent(["comps"])
   comps_answer, _ = comps_agent.run(
       "Per the comps talent, compute EV, EV/Revenue, EV/EBITDA and P/E "
       "(all NTM) for these friends with run_python:n" + PEERS +
       "nThen print the total comps desk plus min/twenty fifth/median/seventy fifth/max "
       "abstract stats as JSON below the marker COMPS_JSON: with keys "
       "'desk' (checklist 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.hundreds(m.group(1))
       desk = pd.DataFrame(payload["table"])
       stats = pd.DataFrame(payload["stats"])
       xlsx = "outputs/comps_analysis.xlsx"
       with pd.ExcelAuthor(xlsx, engine="openpyxl") as xl:
           desk.to_excel(xl, sheet_name="Comps", index=False)
           stats.to_excel(xl, sheet_name="Summary Stats")
       from openpyxl import load_workbook
       from openpyxl.types import Font, PatternFill
       wb = load_workbook(xlsx)
       for ws in wb.worksheets:
           for cell in ws[1]:
               cell.font = Font(daring=True, shade="FFFFFF")
               cell.fill = PatternFill("stable", start_color="1F4E79")
           for col in ws.columns:
               w = max(len(str(c.worth)) for c in col if c.worth is just not None)
               ws.column_dimensions[col[0].column_letter].width = w + 3
       wb.save(xlsx)
       print("Wrote", xlsx)
       print(desk.to_string(index=False))
besides Exception as e:
   print("Demo B skipped:", e)

print("n" + "="*76 + "nDEMO C — IC MEMO (ic-memo talent)n" + "="*76)
strive:
   ic_agent = SkillAgent(["ic-memo"])
   ic_answer, _ = ic_agent.run(
       "Draft a first-round IC memo per the talent for a hypothetical "
       "buyout of Meridian Software (see information under). Base the valuation "
       "framing on ~11x EV/EBITDA entry, 45% leverage, 5-yr maintain. Use "
       "run_python for any fast math (e.g., tough 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)")
besides 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)
   strive:
       spec = yaml.safe_load(open(path))
       print(json.dumps(spec, indent=2, default=str)[:2500])
       print("nDeploy circulate: resolve file refs -> add expertise -> create "
             "leaf-worker subagents -> POST orchestrator to /v1/brokers "
             "(see scripts/orchestrate.py for the handoff_request occasion loop).")
   besides Exception as e:
       print("Could not parse yaml:", e)
else:
   print("No cookbook yaml discovered on this department — see "
         "managed-agent-cookbooks/ READMEs on GitHub.")
print("n" + "="*76)
print("DONE. Artifacts in ./outputs/:", os.listdir("outputs"))
print("""
Where to go subsequent:
* Swap SAMPLE_CO / PEERS for actual information through the repo's MCP connectors
  (Daloopa, FactSet, S&P, Morningstar, PitchBook...) — subscriptions apply.
* Load different expertise: SkillAgent(["lbo"]), ["merger"], ["earnings"],
  ["rebalance"], ["tlh"], ["kyc"] ... — see `sorted(registry.expertise)`.
* Stack expertise: SkillAgent(["comps", "dcf"]) for a football-field workflow.
* For manufacturing, set up as a Cowork plugin or deploy through Managed Agents
  as a substitute of this Colab loop — similar expertise, ruled runtime.
All outputs are drafts for certified human evaluate — not funding recommendation.
""")