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

In this tutorial, we build an advanced workflow around Anthropic’s financial-services repository and reproduce its skill-driven architecture in pure Python. We begin by installing the required libraries, cloning the repository, and programmatically mapping its agents, vertical plugins, partner integrations, managed-agent cookbooks, and financial analysis skills. We then parse the repository’s SKILL.md files into a searchable registry and construct a reusable SkillAgent that injects selected financial playbooks into the Anthropic Messages API while supporting an iterative tool-use loop for Python calculations and file generation. Using this architecture, we execute a synthetic discounted cash flow valuation, generate a WACC and terminal-growth sensitivity heatmap, perform comparable-company analysis with formatted Excel output, draft a private-equity investment committee memo, and inspect a managed-agent deployment specification without sending a live deployment request.

Copy CodeCopiedUse a different Browser

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)

We install the required Python libraries, clone Anthropic’s financial-services repository, and prepare the Google Colab runtime for execution. We retrieve the Anthropic API key from Colab secrets, environment variables, or a secure interactive prompt. We then initialize the official Anthropic SDK and select the Claude model that powers the financial-analysis workflows.

Copy CodeCopiedUse a different Browser

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

We inspect the repository structure and identify its agent plugins, vertical plugins, partner integrations, managed-agent cookbooks, and available commands. We locate MCP configuration files and display the external financial data connectors defined in the repository. We then parse each SKILL.md file, extract its YAML metadata and methodology, and register every unique skill for searchable access.

Copy CodeCopiedUse a different Browser

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/.", "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

We define tools that allow Claude to execute Python calculations and save generated deliverables inside the Colab environment. We build a persistent Python namespace so numerical models, tables, and intermediate variables remain available across multiple agent turns. We then create the SkillAgent class, inject selected financial playbooks into its system prompt, and manage the Anthropic Messages API tool-use loop.

Copy CodeCopiedUse a different Browser

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)

We provide synthetic operating assumptions and instruct the DCF skill agent to construct a five-year unlevered cash-flow valuation. We use the Python execution tool to calculate enterprise value, equity value, implied share price, and a two-dimensional sensitivity matrix. We then extract the structured sensitivity results and visualize the relationship between WACC, terminal growth, and implied valuation through a heatmap.

Copy CodeCopiedUse a different Browser

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

[truncated for AI cost control]