Финансовый AI-агент на Python: MCP и CodeAct

Это продолжение статьи, в которой мы создавали MCP-сервер на примере Финам. В этой части я покажу, как построить полноценного финансового ИИ-ассистента на Python. Научимся реализовывать MCP-клиента, подключать инструменты к LLM, попробуем разные архитектуры ИИ-агентов (в том числе “темную лошадку” - CodeAct, но об этом позже).

В первой части мы разобрались с концепцией Model Context Protocol (MCP) и создали MCP-сервер для API биржи Finam — он предоставляет инструменты для получения котировок, работы с ордерами и управления портфелем. MCP позволяет ассистентам не просто генерировать текст, но и выполнять реальные действия, сохраняя при этом модульность и удобство поддержки.

Кстати, недавно я также рассказывал, как подключить ранее созданный MCP-сервер к Claude Desktop для личного использования (получение данных бирж, аналитика, инсайты) без всякого кода и это уже работает очень даже классно (загляните обязательно). Но мы не будем останавливаться на этом и пойдем дальше написав собственное ИИ-приложение помощника инвестора.

Идея проста: вы пишете «Покажи топ-10 акций по росту за месяц с объёмом от 50 млн рублей», а ассистент самостоятельно получает данные, анализирует их, создает визуализации и выдает рекомендации. Своего рода "вайбтрейдер" на стероидах. Подобные решения уже появляются на рынке — например, стартап ROCKFLOW.

Описанная архитектура универсальна — те же принципы работают для банковских помощников, туристических агентов, личных ассистентов и любых других доменов. Меняются только промпты и MCP-интеграции, а подход остается тем же.

Давайте приступать!

Версия №1: MCP-агент

Сделаем MCP-агента. Вот как это работает. У нас есть:

• Модель — ChatGPT, Claude, Gemini;

• MCP-сервер - Finam-MCP сервер, то что сделали в прошлой части;

• MCP-клиент - то что пытаемся сделать в этой.

Напомню, что MCP-сервер содержит в себе выполняемые функции tools. Каждый tool содержит в себе полное описание функции как ее применять LLM: имя, описание, входные и выходные значения.

Как это работает по-шагово:

1. С помощью tools/list запроса на MCP-сервер получаем общий список таких описаний, которые мы кладем в системный промпт модели.

2. При каждом обращении пользователя к ИИ-модели, она уже сама будет решать, когда ей вызывать эти функции.

3. Когда она все таки решит их вызывать (испустит специальный токен), то провайдер приостановит модель и вернет tool-message клиенту.

4. Мы на клиенте уже вызовем выбранный(ые) моделью инструмент(ы) с вписанными моделью параметрами.

5. Так как по сути все реализации инструментов находятся на стороне MCP-сервера, то мы вызываем их через эндпоинт tools/call.

6. Сервер выполнит каждый инструмент и выдаст ответ.

7. Этот ответ мы напрямую передаем в контекст модели

8. Модель возвращает свой ответ

9. Возвращаемся к шагу 2

Вот прекрасная схема, которая поясняет все сказанное мною.

На практике для обращений к MCP серверу в клиентском коде мы можем использовать класс Client из библиотеки fastmcp. Нас интересуют его методы list_tools() - получить список всех инструментов сервера и call_tool() - вызвать инструмент с введенными аргументами. Вот базовый пример использования. Перед запуском кода, не забудьте запустить этот MCP-сервер из прошлой статьи.

import asyncio from fastmcp import Client # HTTP MCP-client client = Client("<http://localhost:3000/mcp>") async def main(): async with client: # Базовое взаимодействие с сервером await client.ping() # Список доступных инструментов tools = await client.list_tools() # Выполнить определенный инструмент result = await client.call_tool("market_data_get_last_trades", {"symbol": "SBER@MISX"}) print(result) asyncio.run(main())

Хорошо, сделаем теперь полноценного MCP-агента. В экосистеме langchain (которой я пользуюсь для построения ИИ-агентов) есть удобный модуль langchain-mcp-adapters для бесшовного встраивания MCP инструментов к агенту. Работает это так, что объект MultiServerMCPClient() подключается по заданной конфигурации к MCP-серверам и методом get_tools() выдает все доступные инструменты в виде списка подготовленных langchain tool-объектов. Нам ничего не остается, кроме как подключить эти tool-объекты к любому провайдеру (ChatOpenAI, ChatAnthropic, ChatGoogleGenerativeAI) через метод bind_tools(). Вот как это выглядит:

from langchain_mcp_adapters.client import MultiServerMCPClient from langchain_anthropic import ChatAnthropic client = MultiServerMCPClient( { "finam": { "transport": "streamable_http", "url": f"http://{MCP_SERVER_HOST}:{MCP_SERVER_PORT}/mcp", "headers": {"FINAM-API-KEY": FINAM_API_KEY, "FINAM-ACCOUNT-ID": FINAM_ACCOUNT_ID} }, } ) tools = await client.get_tools() agent = ChatAnthropic("claude-sonnet-4-5").bind_tools(tools) agent.invoke("Покажи мне топ-10 финансовых акций по росту за месяц с объёмом от 50 млн рублей")

Не забываем указывать через переменные MCP_SERVER_HOST хост и порт MCP_SERVER_PORT сервера, а также авторизационные заголовки FINAM_API_KEY и FINAM_ACCOUNT_ID, которые можно получить в личном кабинете Финам.

Готово! Можем тестировать. Накидал базовый интерфейс в streamlit приложения “Вайбтрейдер”. Проверочный запрос у меня такой: “Покажи мне топ-10 финансовых акций по росту за месяц с объёмом от 50 млн рублей”. В этом случае агенту нужно получить список доступных финансовых инструментов методом get_assets, отфильтровать, проанализировать и сравнить их и на основе этого дать свои рекомендации.

Ввожу промпт в чате, ассистент принял запрос, вызвал нужный первый инструмент, получил данные и ... вышел из чата.

Дело в том, что запрос выдает сырых данных о 8500 доступных в Финам инструментов общим объемом более полумиллиона токенов (580 тыс. токенов). Для справки, Claude Sonnet 4.5 может только обработать 200 тыс., а GPT-5.1 - 1 миллион. Не говоря уже о том, сколько это будет стоить и как долго будет обрабатываться.

Итак, явная проблема базового похода - чрезмерное потребление токенов контекстного окна инструментами. Это снижает эффективность агентов. Токены потребляются вначале при определении инструментов и как промежуточные результаты (все идет в контекстное окно ИИ-модели). А что если мы захотим добавить больше интеграций - получать курсы валют, криптовалют?

Помимо этого инструменты выполняются последовательно, без всякой асинхронности и обработки ошибок, а получаемые данные могут содержать персональную информацию

В целом, в вашем случае если у вас немного инструментов проблема переполнения токенами может не возникать и вы можете остановится на этой простой реализации. Но в случае финансового ИИ-агента с необходимостью обрабатывать большие данные проблема критична и нам нужна более эффективная агентская архитектура. И такая есть.

CodeAct

Решение приходит от Anthropic и Cloudflare: вызывать инструменты LLM не через MCP, а через код - созданную API обертку над MCP. Не пытаться заставить LLM пользоваться выдуманной нами абстракцией с инструментами, которые она едва видела при обучении, а более нативными и понятными для нее языками программирования. А программирует LLM достаточно хорошо (вспомнить что весь 2025 год стал годом вайбкодинга), так почему бы не применить ее способности для наших целей?

Вот как это работает. Вместо того чтобы напрямую передавать инструменты в LLM мы пишем на их основе специальные кодовые API-модули, которые уже даем на вход модели. Она пишет свой код с вызовом этих API-модулей. Код выполняется в безопасной изолированной песочнице. Каждый раз, когда в коде происходит вызов API он проходит через наше приложение используя binding-объект. Там уже происходит непосредственное обращение к MCP-серверу и возврат результатов обратно в код.

Схематично, в сравнение с обычным подходом, это можно представить так:

Такой подход дает сразу несколько преимуществ.

• Постепенное раскрытие возможностей. Агенту не сразу подаются на вход все возможности интеграций. Он может их раскрывать постепенно, по мере надобности. Это сильно экономит токены на старте и поддерживает чистоту в контекстном окне.

• Эффективное обработка больших объемов данных. Вместо 8500 инструментов, агент может отфильтровать и показать только 10. Все без перегрузки контекстного окна (как случилось в нашем случае).

• Более гибкий и настраиваемый контроль потока. Агент может писать условия, циклы, обработку ошибок, выполнять операции асинхронно, писать callback. Безграничный простор программистской фантазии.

• Сохранение конфиденциальности. В binding-объекте можно настроить автоматическую замену чувствительных данных (email, телефоны, имена пользователей), если агент с таковыми работает.

• Сохранение состояния и навыков. Все переменные и функции написанные агентом между разными вызовами сохраняются. А значит их можно переиспользовать - полученные данные не запрашивать снова, а написанные функции использовать как готовые модули - skills.

Если говорить в более общем плане, то что мы с вами сделали в первой версии называется ReAct (“Reasoning + Acting”) - агент думает, выполняет внешние действия, получает обратную связь и снова думает, повторяя цикл. “Выполняет действия” через данные ему функции путем испускания специального токена с json/text форматом. Но обычно никто не говорит ReAct-агент, а просто ИИ-агент. В статье я упоминаю его как MCP-агент, потому что в качестве этих функций “Acting” выступают MCP инструменты.

Новый подход же обрел название CodeAct, когда действия агента происходят через написанный им код. CodeAct был предложен исследователями из Apple еще в начале 2024 года, а сейчас например реализован в агенте Manus.io. Реализуем же мы его тоже!

Версия №2: CodeAct-Агент

Итак по сути нам нужно реализовать

1. Автоматизированную генерацию API кода на базе MCP;

2. Binding-объект для возможности вызова MCP изнутри песочницы;

3. Песочницу - безопасную среду выполнения агентского кода;

4. Дополнительные инструменты для ИИ-агента.

Давайте все по порядку. Прежде чем что-то писать, а на каком языке будет писать ассистент? Нужен простой, но мощный скриптовый язык, при этом достаточно популярный чтобы оказаться в обучающих данных LLM. Здесь сразу два фаворита: Python и JS. Оба ИИ-модели знают хорошо и применяют в продакшене (ChatGPT - Python, Claude - JS). Я остановился на моем любимом Python).

Но без субъективщины скажу, что разрабатываемый ИИ-ассистент должен делать много аналитики с финансовыми данными. На Python много пишут аналитики и дата сайнтисты. А LLM были обучены как раз на их открытых ноутбуках в гитхаб. Кстати еще из плюсов что Питон изящный и компактный язык, соответственно потребляет меньше токенов. Хотя это всего лишь мое мнение и никаких таких исследований я не видел. Если не согласны - приглашаю к обсуждению в комментариях.

Но должен признать, что код на JS может быть быстрее. И вообще забегая вперед, скажу что если бы писал ИИ-ассистент писал на JS было бы проще с контейнеризацией и безопасностью. Но это уже проблемы будущего меня. А пока достаем Python-интерпретатор).

Этап 1: Генерация API-обертки

Итак, убираем все langchain-mcp-adapters и вспоминаем про реализацию MCP-клиента Client из модуля fastmcp. После получения списка инструментов методом list_tools() превращаем их в код используя реализованную функцию mcp_to_code().

import asyncio from fastmcp import Client from mcp import Tool from config import settings mcp_config = { "mcpServers": { "finam": { "transport": "http", "url": f"{settings.MCP_SERVER_URL}/mcp" }, } } async def get_tool_list() -> list[Tool]: """Получение списка доступных tools""" async with Client(mcp_config) as client: return await client.list_tools() def generate_mcp_files(): """Сгенерировать API обертку""" tools = asyncio.run(get_tool_list()) mcp_to_code(tools)

Полную реализацию mcp_to_code я показывать не буду, так как она достаточно большая и вы можете посмотреть ее сами в исходном коде. Скажу лишь про алгоритм. Во-первых мы сохраняем кэш сигнатуры инструментов, чтобы при каждом перезапуске не генерировать один и тот же код. Во-вторых нам нужно в сгенерированном коде как-то доходчиво при этом кратко объяснить модели как этим кодом пользоваться. Помимо обычных docstring использую Pydantic модели.

Возьмем для примера assets_get_asset - получение данных по заданной ценной бумаге. Из полезных данных в объекте Tool получаемом с MCP-сервера содержится:

• tool.name - Имя инструмента (assets_get_asset),

• tool.description - Описание инструмента (”””Получение информации по конкретному инструменту”””)

• tool.meta - Мета теги инструмента ({”tags”: “assets”})

• tool.inputSchema - входная схема

• tool.outputSchema - выходная схема

Алгоритм создания будет таким

1. Формируем папку и создаем в ней .py файл по имени инструмента (tool.name) и его тегу (tool.meta) (assets_get_asset → assets/get_asset.py)

2. Сверху файла у нас будет docstring в виде описания tool.description

3. Далее по входной и выходной схеме (tool.inputSchema и tool.outputSchema) генерируем pydantic модели. В этом поможет модуль datamodel_code_generator, который из json_schema сгенерирует все классы с необходимыми импортами. Сгенерированный код записываем в файл.

4. Формируем сигнатуру функции по шаблону: имя инструмента + входная pydantic модель + выходной pydantic модель (def get_asset(symbol: str) -> AssetsGetOutput:) — записываем в файл.

5. Внутри функции ставим: знак табуляции (\t) + return + имя выходной pydantic модели + метод model_validate + mcp.call_tool + имя инструмента + входные параметры. Записываем в файл

В итоге сгенерированный код получается примерно таким:

"""Получение информации по конкретному инструменту (лот, шаг цены, дата экспирации фьючерса)""" from __future__ import annotations from pydantic import BaseModel class FinamDecimal(BaseModel): value: str | None = '0.0' class AssetsGetOutput(BaseModel): id: str ticker: str mic: str isin: str type: str name: str board: str decimals: int min_step: str lot_size: FinamDecimal expiration_date: str | None = None def get_asset(symbol: str) -> AssetsGetOutput: return AssetsGetOutput.model_validate(mcp.call_tool("assets_get_asset", symbol=symbol))

Что за mcp.call_tool? Разбираем дальше.

Этап 2: Binding-объект

Сейчас при обычной попытке запуска API оболочки код выйдет с ошибкой - NameError: name 'mcp' is not defined. Все потому что, mcp - это binding-объект, вызывающий MCP. Его реализация ИИ-агенту показываться не будет, но в глобальное пространство имен Python сессии в песочнице он добавлен будет. Выглядит он так

class MCPBinding: """Binding для доступа к MCP серверу из sandbox""" def __init__(self, finam_api_key: str, finam_account_id: str): config = mcp_config.copy() if finam_api_key and finam_account_id: config["mcpServers"]["finam"]["headers"] = {"FINAM-API-KEY": finam_api_key, "FINAM-ACCOUNT-ID": finam_account_id} self._client = Client(config) def call_tool(self, name: str, **kwargs): """Вызов MCP tool (для llm)""" return asyncio.run(self.async_call_tool(name, **kwargs)) async def async_call_tool(self, name: str, **kwargs) -> dict: """Вызов MCP tool""" async with self._client: return (await self._client.call_tool(name, kwargs)).structured_content

В инициализаторе задаю авторизационные заголовки для MCP-сервера и создаю объект Client. Реализую два метода вызова инструментов - синхронную и асинхронную.

У этого кода с первого этапа есть проблема - используется синхронный call_tool для вызова инструментов. То есть каждый вызов инструмента приложение будет терпеливо ждать его полного выполнения, а не делать в фоне. Конечно можно (и нужно!) использовать асинхронную функцию. Но вы не представляете сколько это создает проблем в Python (которых, кстати, в JS не возникает).

Стоило сделать API обертку асинхронной, расставив async и await в функциях в нужных местах, как ИИ-агент писал код, учитывающий асинхронность и честно упаковывал его в asyncio.run(). И никакие уговоры (”пиши код как обычный синхронный, он выполнится, не надо ставить asyncio.run()!”) не срабатывали, что вызывало кучу ошибок. Поэтому оставляю читателю возможность оптимизировать процесс самостоятельно.

Да и вообще основная задача максимально сделать вид для ассистента, что весь его код выполнится в “интерактивной Python консоли” или в “Jupyter ноутбуке”. Кстати о запуске кода.

Этап 3: Запуск кода

Да, а вот последствия выбора Python в качестве главного языка. Если бы выбрал JS можно было без проблем использовать изолированные легковесные окружения V8 JavaScript engine. Но работаем с теми решениями, которые есть в инфраструктуре Python, и в таблице я привожу основные варианты и их свойства.

• Statefull - означает, что состояние (переменные и функции) будут сохраняться в рамках сессии между разными запусками кода.

• Время запуска - сколько нужно времени чтобы запустить код.

• Изоляция среды - может ли написанный агентом код получать переменные среды, читать/записывать файлы, делать запросы в сеть, запускать подпроцессы или FFI.

• Десериализация Python объектов - сможем ли мы взять результат выполнения и использовать полноценно в своем коде. Например для визуализации пользователю (таблица, график).

• Импорт сторонних Python модулей - возможность импортировать внешние модули, в том числе нашу API-обертку.

Я остановился на варианте со встроенной в Python функцией exec(). У llamaIndex есть пример реализации, его и взял. Вариант самый простой, но самый не безопасный. В целом в статье хочется проверить и донести идею, но...

Итак, мы подготовили рабочее место для агента. Пора его самого сделать

Этап 4. Построение агента

Для агента реализуем следующие инструменты:

• Вызов кода - run_python_code(code: str) -> tuple[str, list[tuple[str, Any]] | None]

• Список доступных инструментов - list_tool(path: str = "/") -> str

• Прочитать код специфичного инструмента(ов) - read_tool(tool_name: str | list[str]) -> str

В оригинальном исследовании CodeAct бралась небольшая модель и дообучалась во время общения писать выполняемый код для решения задачи. В целом, передовые модели и так достаточно умны, их не нужно дообучать под пользование инструментами выполнения кода. Поэтому для модели я создал инструмент run_python_code, который получает весь код от модели в качестве входного аргумента и возвращает результат выполнения. Этот инструмент также возвращает в код приложения артефакт - возвращаемый Python объект. Использую для визуализации пользователю.

Не забываем что в среду выполнения нужно также встроить binding-объект. Делаю это через Python прием closure (замыкание) в паре с паттерном фабрика (когда функция create_code_executor_tool создает инструмент run_python_code). Код:

def create_code_executor_tool(finam_api_key: str, finam_account_id: str): """Создаёт tool для выполнения кода с MCP binding""" # Создаём binding один раз mcp_binding = MCPBinding(finam_api_key, finam_account_id) code_executor = SimpleCodeExecutor( locals={}, globals={"mcp": mcp_binding} ) @tool(response_format="content_and_artifact") def run_python_code(code: str) -> tuple[str, list[tuple[str, Any]] | None]: """Run Python code (supports async/await)""" success, exec_output, return_value = code_executor.execute(code) result = '✅' if success else '❌' if exec_output: result += ' ' + exec_output artifacts = None if return_value is not None: # Сериализуем результаты для возможности сохранения output = return_value if isinstance(return_value, tuple) else [return_value] artifacts = [serialize_output(elem) for elem in output] return result, artifacts return run_python_code

Далее инструменты list_tool и read_tool. Они нужны для того самого постепенного раскрытия возможностей ИИ-моделью и запроса информации об инструментах по требованию. list_tools дает список имен инструментов и их описаний, read_tool - полный Python код API-обертки. Вот их реализация:

@tool def list_tool(path: str = "/") -> str: """List of short descriptions of all tools in the specified directory""" target_dir = settings.MCP_SERVERS_DIR / path.lstrip("/") if not target_dir.exists() or not target_dir.is_dir(): return f"Directory {path} not found" descriptions = [] for py_file in sorted(target_dir.rglob("*.py")): if "__pycache__" in str(py_file) or py_file.name == "__init__.py": continue desc = get_short_tool_description(py_file) if desc: descriptions.append(desc) return "\\n\\n".join(descriptions) if descriptions else "No tools found" @tool def read_tool(tool_name: str | list[str]) -> str: """Get detailed description of one or more tools to understand their interface""" tool_names = [tool_name] if isinstance(tool_name, str) else tool_name results = [] for name in tool_names: clean_name = name.removesuffix(".py") normalized_name = clean_name.replace("/", ".") parts = normalized_name.split(".") file_path = settings.MCP_SERVERS_DIR / "/".join(parts[:-1]) / f"{parts[-1]}.py" if not file_path.exists(): results.append(f"Tool {name} not found") continue try: content = file_path.read_text(encoding="utf-8") results.append(f"=== {name} ===\\n{content}") except Exception as e: results.append(f"Error reading {name}: {e}") return "\\n\\n".join(results)

И последний но не менее важный штрих - системный промпт. Помимо описания инструкций, предметной области, примеров кода, добавлю в него еще file-tree доступных серверов для общей ориентации по файлам API-библиотеки. Выглядит оно примерно так:

└── finam ├── order │ ├── get_order.py │ ├── get_orders.py │ ├── cancel_order.py │ └── place_order.py ├── market_data │ ├── get_bars.py │ ├── get_last_trades.py │ ├── get_last_quote.py │ └── get_order_book.py ├── account │ ├── get_trades.py │ ├── get_transactions.py │ └── get_account_info.py └── assets ├── get_asset.py ├── get_assets.py ├── get_schedule.py ├── get_exchanges.py ├── get_options_chain.py └── get_params.py

Вобщем, все готово. Полный код выложил в гитхаб.

Тестирую в том же streamlit приложении на том же входном промпте. В этот раз использовал Claude-Sonnet-4.5 модель, так как она более креативная на написание кода. После запуска агент изучил предоставленные ему MCP, вызвал их в своем коде, обработал данные, а потом красиво это все презентовал через интерактивные plotly графики. Ну лепота!

Выводы

Итак, в результате обновленного подхода CodeAct к выполнению инструментов через код агент получил больше возможностей и стал справляться с серьезными задачами по финансовому анализу. Полная версия кода выложена на github.

В целом, вам может это все и не понадобится и вы остановитесь на обычном MCP-агенте. Привожу таблицу разницы двух подходов:

Если вас как и меня зацепила CodeAct архитектура, то вот вам пару задач со звездочкой в продолжении развития идеи на самостоятельное изучение:

• Асинхронная API обертка MCP инструментов

• Безопасная песочница - выполнение кода через изолированные окружения (Linux контейнеры или Jupyter)

• Сделать human-in-the-loop подтверждения. При создании MCP-сервера мы оставили задел под это в виде мета-тега meta={"sensitive": True}. Можно придумать, как валидировать код агента на наличие этих “чувствительных” инструментов.

А на этом у меня все. Надеюсь было полезно и интересно)

Присоединяйтесь к сообществу LLM разработчиков: @llm_studio

Пробуйте, экспериментируйте, изучайте новое! До новых встреч!