En este tutorial, creamos un flujo de trabajo avanzado en torno al repositorio de servicios financieros de Anthropic y reproducimos su arquitectura basada en habilidades en Python puro. Comenzamos instalando las bibliotecas necesarias, clonando el repositorio y mapeando mediante programación sus agentes, complementos verticales, integraciones de socios, libros de recetas de agentes administrados y habilidades de análisis financiero. Luego analizamos los archivos SKILL.md del repositorio en un registro con capacidad de búsqueda y construimos un SkillAgent reutilizable que inyecta guías financieras seleccionadas en la API de Anthropic Messages mientras admite un bucle iterativo de uso de herramientas para cálculos de Python y generación de archivos. Utilizando esta arquitectura, ejecutamos una valoración sintética de flujo de efectivo descontado, generamos un WACC y un mapa de calor de sensibilidad de crecimiento terminal, realizamos análisis de empresas comparables con resultados formateados en Excel, redactamos un memorando del comité de inversión de capital privado e inspeccionamos una especificación de implementación de agente administrado sin enviar una solicitud de implementación en vivo.
importar subproceso, 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") importar pandas como pd importar yaml importar matplotlib.pyplot como plt REPO_URL = "https://github.com/anthropics/financial-services.git" REPO_DIR = "servicios-financieros" si no os.path.isdir(REPO_DIR): sh(f"git clone –profundidad 1 {REPO_URL} {REPO_DIR}") else: print("Repo ya clonado – saltando.") def get_api_key(): prueba: desde google.colab importar datos de usuario k = userdata.get("ANTHROPIC_API_KEY") if k: devuelve k excepto Excepción: pasa si os.environ.get("ANTHROPIC_API_KEY"): devuelve os.environ["ANTHROPIC_API_KEY"] from getpass import getpass return getpass("Ingrese su clave API de Anthropic: ") os.environ["ANTHROPIC_API_KEY"] = get_api_key() import anthropic client = anthropic.Anthropic() MODEL = "claude-sonnet-4-6" print("SDK listo. Modelo:", MODELO)
Instalamos las bibliotecas de Python necesarias, clonamos el repositorio de servicios financieros de Anthropic y preparamos el tiempo de ejecución de Google Colab para su ejecución. Recuperamos la clave API de Anthropic de los secretos de Colab, variables de entorno o un mensaje interactivo seguro. Luego inicializamos el SDK oficial de Anthropic y seleccionamos el modelo Claude que impulsa los flujos de trabajo de análisis financiero.
def repo_map(raíz=REPO_DIR): filas =[]para tipo, patrón en [ ("agent", f"{root}/plugins/agent-plugins/*"), ("vertical",f"{root}/plugins/vertical-plugins/*"), ("partner", f"{root}/plugins/partner-built/*"), ("cookbook",f"{root}/managed-agent-cookbooks/*"), ]: para p en sorted(glob.glob(pattern)): si no os.path.isdir(p): continuar 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) para f en mcp_files[:1]: print(f"n=== CONECTORES MCP ({f}) ===") intente: cfg = json.load(open(f)) para nombre, srv en cfg.get("mcpServers", cfg).items(): print(f" {name:<14} -> {srv.get('url', srv)}") excepto excepción como e: print(" (no se pudo analizar:", e, ")") FRONTMATTER = re.compile(r"^—s*n(.*?)n—s*n", re.S) clase Habilidad: def __init__(self, ruta): self.path = ruta raw = open(ruta, codificación="utf-8", errores="replace").read() m = FRONTMATTER.match(raw) meta = {} if m: intente: meta = yaml.safe_load(m.group(1)) o {} excepto Excepción: meta = {} self.name = str(meta.get("name") o 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"" clase SkillRegistry: def __init__(self, root=REPO_DIR): rutas = sorted(glob.glob(f"{root}/plugins/vertical-plugins/**/SKILL.md", recursive=True)) rutas += sorted(glob.glob(f"{root}/plugins/**/SKILL.md", recursive=True)) self.skills = {} para p in rutas: s = Skill(p) self.skills.setdefault(s.name.lower(), s) def buscar(self, query): q = query.lower() visitas = [s para k, s en self.skills.items() si q en k] si no visitas: visitas = [s para s en self.skills.values() si q en s.description.lower()] devolver visitas def get(self, query): hits = self.find(query) si no hits: rise KeyError(f"No hay habilidad que coincida con '{query}'. " f"Disponible: {sorted(self.skills)[:40]}") devolver hits[0]registro = SkillRegistry() print(f"nLoaded {len(registry.skills)} habilidades únicas.") print("Muestra:", ordenado(registry.skills)[:12], "…")
Inspeccionamos la estructura del repositorio e identificamos los complementos de sus agentes, complementos verticales, integraciones de socios, libros de recetas de agentes administrados y comandos disponibles. Localizamos los archivos de configuración de MCP y mostramos los conectores de datos financieros externos definidos en el repositorio. Luego analizamos cada archivo SKILL.md, extraemos sus metadatos y metodología YAML y registramos cada habilidad única para acceso con capacidad de búsqueda.
os.makedirs("outputs", exist_ok=True) TOOLS = [ { "name": "run_python", "description": ("Ejecute el código Python y devuelva stdout. pandas como pd " "y numpy como np están preimportados. Utilice print() para " "devolver resultados. El estado persiste en todas las llamadas."), "input_schema": { "type": "object", "properties": {"code": {"type": "string"}}, "required": ["code"], }, }, { "name": "save_file", "description": "Guardar contenido de texto en salidas/.", "input_schema": { "type": "object", "properties": {"filename": {"type": "string"}, "content": {"type": "string"}}, "required": ["filename", "content"], }, }, ] _PY_NS = {} def _tool_run_python(código): importar numpy como np _PY_NS.setdefault("pd", pd); _PY_NS.setdefault("np", np) buf = io.StringIO() intente: con contextlib.redirect_stdout(buf): exec(code, _PY_NS) out = buf.getvalue() return out[:6000] if out else "(no stdout – use print())" excepto Excepción como e: return f"ERROR: {type(e).__name__}: {e}" def _tool_save_file(nombre de archivo, contenido): seguro = os.path.basename(nombre de archivo) ruta = os.path.join("outputs", seguro) open(ruta, "w", codificación="utf-8").write(content) return f"Guardado {ruta} ({len(content)} caracteres)" DISPATCH = {"run_python": lambda i: _tool_run_python(i["code"]), "save_file": lambda i: _tool_save_file(i["filename"], i["content"])} BASE_SYSTEM = """Usted es un asistente de analista financiero que opera con los manuales de habilidades que se proporcionan a continuación (del repositorio de servicios financieros de Anthropic). Siga de cerca la metodología, las convenciones y el formato de salida de la habilidad. Utilice la herramienta run_python para todo el trabajo numérico; nunca haga aritmética mentalmente. Utilice save_file para entregables finales. Todo el trabajo es un BORRADOR para revisión humana; no lo presente como consejo de inversión.""" class SkillAgent: """Reproducción mínima de la activación de habilidades de Cowork: las habilidades elegidas se concatenan en el indicador del sistema; luego, el agente ejecuta un ciclo de uso de herramientas estándar en la API de Mensajes hasta que el modelo se detiene.""" def __init__(self, skills_queries, max_skill_chars=12000, verbose=True): self.skills = [registry.get(q) para q en skills_queries] bloques =[]para s en self.skills: blocks.append(f"nn===== HABILIDAD: {s.name} =====n" f"{s.description}n{s.body[:max_skill_chars]}") self.system = BASE_SYSTEM + "".join(blocks) self.verbose = detallado def run(self, Prompt, max_turns=12): mensajes = [{"role": "user", "content": Prompt}] para el rango de turnos (max_turns): resp = client.messages.create( model=MODEL, max_tokens=8000, system=self.system, tools=TOOLS, message=messages) message.append({"role": "assistant", "content": resp.content}) if resp.stop_reason != "tool_use": final = "".join(b.text para b en resp.content si b.type == "text") devuelve final, resultados de mensajes =[]para bloque en resp.content: if block.type == "tool_use": if self.verbose: print(f" [turn {turn}] herramienta: {block.name}") out = DISPATCH[block.name](block.input) results.append({"type": "tool_result", "tool_use_id": block.id, "content": str(out)}) message.append({"role": "user", "content": results}) return "(presiona max_turns)", mensajes
Definimos herramientas que permiten a Claude ejecutar cálculos de Python y guardar los entregables generados dentro del entorno Colab. Creamos un espacio de nombres Python persistente para que los modelos numéricos, tablas y variables intermedias permanezcan disponibles en múltiples turnos de agentes. Luego creamos la clase SkillAgent, inyectamos guías financieras seleccionadas en el indicador del sistema y administramos el ciclo de uso de herramientas de la API de Anthropic Messages.
SAMPLE_CO = """ Objetivo: 'Meridian Software' (sintético). Datos reales del año fiscal 2025, millones de dólares: ingresos 850 (aumentaron un 18 % interanual) | Margen EBITDA 27 % | D&A 4 % de rev CapEx 5 % de rev | Cambio de NWC 1 % de crecimiento de rev | Tasa impositiva 24 % Deuda neta 320 | Acciones diluidas 92 mm Supuestos: crecimiento de ingresos se desvanece 18% -> 6% linealmente en 5 años, el margen EBITDA se expande 100 puntos básicos en total, WACC 9,5%, crecimiento terminal 2,5% """ print("n" + "="*76 + "nDEMO A — DCF (dcf-model skills)n" + "="*76) intente: dcf_agent = SkillAgent(["dcf"]) dcf_answer, _ = dcf_agent.run( "Ejecute un DCF sin apalancamiento de 5 años según el manual de habilidades de esta empresa:n" + SAMPLE_CO + "nCalcule el valor empresarial, el valor accionario y el precio " "implícito de las acciones con run_python. Luego imprima una cuadrícula de sensibilidad WACC (8,5%-10,5%, pasos de 50 pb) x " "crecimiento terminal (1,5%-3,5%, pasos de 50 pb) de " "implica el precio de la acción como un objeto JSON bajo el marcador SENS_JSON: y proporciona " "un resumen conciso.") 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, errores="coerce") fig, ax = plt.subplots(figsize=(7, 4)) im = ax.imshow(sens.values, cmap="RdYlGn", aspecto="auto") ax.set_xticks(range(len(sens.columns)), sens.columns) ax.set_yticks(range(len(sens.index)), sens.index) ax.set_xlabel("Crecimiento terminal"); ax.set_ylabel("WACC") ax.set_title("Sensibilidad implícita del precio de las acciones ($)") para i en el rango(sens.shape[0]): para j en rango (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() excepto Excepción como e: print("Demostración A omitida:", e)
Proporcionamos supuestos operativos sintéticos e instruimos al agente experto del DCF para que construya una valoración del flujo de efectivo no apalancado a cinco años. Utilizamos la herramienta de ejecución Python para calcular el valor empresarial, el valor de las acciones, el precio implícito de las acciones y una matriz de sensibilidad bidimensional. Luego extraemos los resultados de sensibilidad estructurados y visualizamos la relación entre WACC, crecimiento terminal y valoración implícita a través de un mapa de calor.
PEERS = """ Conjunto de pares sintéticos ($mm excepto por acción): Ticker Precio Acciones Deuda neta 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 (habilidad de análisis de composiciones) -> Exceln" + "="*76) intente: comps_agent = SkillAgent(["comps"]) comps_answer, _ = comps_agent.run( "Según la habilidad de comparaciones, calcule EV, EV/Ingresos, EV/EBITDA y P/E " "(todos NTM) para estos pares con run_python:n" + PEERS + "nLuego imprima la tabla de comparaciones completa más las estadísticas de resumen min/25/mediana/75/max " "como JSON bajo el marcador COMPS_JSON: con las claves " "'tabla' (lista de dictados de filas) y 'estadísticas' (dictado de dictados).") print("n— NARRATIVA COMPS —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" con pd.ExcelWriter(xlsx, motor="openpyxl") como xl: table.to_excel(xl,sheet_name="Comps", index=False) stats.to_excel(xl,sheet_name="Summary Stats") de openpyxl importar load_workbook desde openpyxl.styles importa fuente, PatternFill wb = load_workbook(xlsx) para ws en wb.worksheets: para celda en ws[1]: cell.font = Font(bold=True, color="FFFFFF") cell.fill = PatternFill("solid", start_color="1F4E79") para col en ws.columns: w = max(len(str(c.value)) para c en col si c.value no es Ninguno) ws.column_dimensions[col[0].column_letter].width = w + 3 wb.save(xlsx) print("Escribió", xlsx) print(table.to_string(index=False)) excepto Excepción como e: print("Demostración B omitida:", e)
Proporcionamos un conjunto de pares sintéticos y calculamos el valor empresarial, la relación EV-ingresos, EV-EBITDA y múltiplos precio-beneficio. Convertimos la respuesta JSON estructurada del agente en DataFrames detallados de estadísticas resumidas y de empresas comparables. Luego exportamos el análisis a un libro de Excel de varias hojas y aplicamos un formato de encabezado profesional y un tamaño de columna automático.
print("n" + "="*76 + "nDEMO C — IC MEMO (habilidad ic-memo)n" + "="*76) try: ic_agent = SkillAgent(["ic-memo"]) ic_answer, _ = ic_agent.run( "Elabore un memo IC de primera ronda según la habilidad para una " "compra hipotética de Meridian Software (consulte los datos a continuación). Base la valoración " "en el marco ~11x entrada EV/EBITDA, 45% de apalancamiento, retención de 5 años Utilice " "run_python para cualquier cálculo rápido (por ejemplo, cálculos aproximados de MOIC/IRR) y " "guarde la nota con save_file como ic_memo_meridian.md.n" + SAMPLE_CO) print("n— IC MEMO (primeros 1500 caracteres de respuesta) —n", ic_answer[:1500]) memo_path = "outputs/ic_memo_meridian.md" if os.path.exists(memo_path): print(f"nMemo guardado -> {memo_path} " f"({os.path.getsize(memo_path)} bytes)") excepto excepción como e: print("Demo C omitida:", e) print("n" + "="*76 + "nPARTE 7: LIBRO DE COCINA DEL AGENTE ADMINISTRADO (ejecución en seco)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: ruta = yamls[0]print("Inspeccionando:", ruta) try: spec = yaml.safe_load(open(path)) print(json.dumps(spec, indent=2, default=str)[:2500]) print("nFlujo de implementación: resolver referencias de archivos -> cargar habilidades -> crear " "subagentes de trabajadores de hoja -> POST orquestador en /v1/agents " "(consulte scripts/orchestrate.py para obtener más información handoff_request event loop).") excepto excepción como e: print("No se pudo analizar yaml:", e) else: print("No se encontró ningún libro de cocina yaml en esta rama; consulte " "libros de cocina de agente administrado/README en GitHub.") print("n" + "="*76) print("HECHO. Artefactos en ./outputs/:", os.listdir("outputs")) print(""" Dónde ir a continuación: * Cambie SAMPLE_CO / PEERS por datos reales a través de los conectores MCP del repositorio (Daloopa, FactSet, S&P, Morningstar, PitchBook…) — se aplican suscripciones * Cargue otras habilidades: SkillAgent(["lbo"]), ["merger"], ["earnings"], ["rebalance"], ["tlh"], ["kyc"]… — consulte. `sorted(registry.skills)`. * Habilidades de pila: SkillAgent(["comps", "dcf"]) para un flujo de trabajo de campo de fútbol. * Para producción, instálelo como un complemento de Cowork o impleméntelo a través de agentes administrados en lugar de este bucle de Colab: las mismas habilidades, tiempo de ejecución gobernado. Todos los resultados son borradores para revisión humana calificada, no asesoramiento de inversión.
Aplicamos la habilidad del memorando del comité de inversiones a una compra hipotética de software y calculamos las métricas de retorno de respaldo con Python. Guardamos la nota resultante como un entregable de Markdown y verificamos que el archivo generado exista en el directorio de salida. Luego inspeccionamos un libro de recetas de agente administrado, mostramos la especificación de implementación y concluimos revisando los artefactos generados y las posibles extensiones de producción.
Al completar este tutorial, implementamos una aproximación práctica basada en Colab de la capacidad de servicios financieros y el marco de agentes de Anthropic, preservando al mismo tiempo el enfoque metodológico del repositorio para el análisis financiero. Combinamos el descubrimiento estructurado de habilidades, la construcción dinámica de indicaciones del sistema, la ejecución persistente de Python, la orquestación de herramientas basadas en API y la generación automatizada de entregables en un único flujo de trabajo reutilizable. También demostramos cómo la misma arquitectura de agente admite múltiples casos de uso financiero, incluida la valoración DCF, el análisis de compensaciones comerciales, las pruebas de sensibilidad, los informes de Excel y la preparación de memorandos del comité de inversiones. A partir de aquí, ampliamos el sistema cargando habilidades adicionales, combinando múltiples manuales de valoración, conectándonos a proveedores de datos financieros con licencia a través de integraciones de MCP y reemplazando el entorno de pruebas del tutorial con un tiempo de ejecución de producción gobernado en Claude Code, Cowork o Agentes administrados.
Consulte el código completo aquí. Además, no dude en seguirnos en Twitter y no olvide unirse a nuestro SubReddit de más de 150.000 ml y suscribirse a nuestro boletín. ¡Esperar! estas en telegrama? Ahora también puedes unirte a nosotros en Telegram.
¿Necesita asociarse con nosotros para promocionar su repositorio de GitHub O su página principal de Hugging O su lanzamiento de producto O seminario web, etc.? Conéctate con nosotros
A Sana Hassan, pasante de consultoría en Marktechpost y estudiante de doble titulación en IIT Madras, le apasiona aplicar la tecnología y la inteligencia artificial para abordar los desafíos del mundo real. Con un gran interés en resolver problemas prácticos, aporta una nueva perspectiva a la intersección de la IA y las soluciones de la vida real.
