我提供了 McpAgent 以支持基于 MCP 协议的 tool call。以下是使用 McpAgent 接入 mcp.stdio_client 的简要入门。

点击这里开始使用 ceo-py

与 MCP 协议一同工作

  1. 引入所需依赖

    • McpAgent 允许你实例化一个能够访问 MCP Tools 的智能体。

    • StdioMcpConfig 是 mcp.client.stdio.StdioServerParameters 的别名，作为 MCP 服务器连接配置

    • @mcp_session(mcp_config: StdioMcpConfig) 允许你将一个函数声明为一个 MCP 会话

    • sync_call 允许你直接调用一个协程函数

     from ceo import (
     McpAgent,
     get_openai_model,
     StdioMcpConfig,
     mcp_session,
     sync_call
 )

  2. 创建一个 MCP 会话

    使用 StdioMcpConfig 创建 MCP 服务器启动与连接配置

     stdio_mcp_config = StdioMcpConfig(
     command='python',
     args=['./my_stdio_mcp_server.py'],
     env=dict(),
     cwd='./mcp_servers'
 )

    被 @mcp_session 修饰的函数，其第一个参数将是 MCP 会话实例，一个函数可以用多个 @mcp_session 修饰，以获取面向不同 MCP 服务器的会话实例

     @sync_call
 @mcp_session(stdio_mcp_config)
 async def run(session, request: str) -> str:
     ...

  3. 在 MCP 会话中创建 McpAgent 实例

    创建 McpAgent 后，你需要调用 fetch_abilities() 方法来向 MCP 服务器获取 tools 配置

     @sync_call
 @mcp_session(stdio_mcp_config)
 async def run(session, request: str) -> str:
     mcp_agent = await McpAgent(session=session, brain=get_openai_model()).fetch_abilities()
     ...

  4. 为 McpAgent 实例分配任务，并等待获取执行结果

     @sync_call
 @mcp_session(stdio_mcp_config)
 async def run(session, request: str) -> str:
     mcp_agent = await McpAgent(session=session, brain=get_openai_model()).fetch_abilities()
     result = await mcp_agent.assign(request).just_do_it()
     return result.conclusion

  5. 运行该函数

     if __name__ == '__main__':
     ret = run(request='What can you do?')
     print(ret)

使用示例

from ceo import (
    McpAgent,
    get_openai_model,
    ability,
    sync_call,
    StdioMcpConfig,
    __BLOG__
)
from ceo.util.mcp_session import mcp_session
from ceo.brain.hook import BeforeActionTaken, AfterActionTaken
from ceo.message import BeforeActionTakenMessage, AfterActionTakenMessage
from dotenv import load_dotenv

load_dotenv()
model = get_openai_model()
stdio_mcp_config = StdioMcpConfig(
    command='python',
    args=['./playwright-plus-python-mcp.py'],
    env=dict(),
    cwd='./mcp_server'
)


@ability(model)
def write_file(filename: str, content: str) -> str:
    with open(filename, 'w', encoding='utf-8') as f:
        f.write(content)
    return f'{content} written to {filename}.'


def before_action_taken(agent: McpAgent, message: BeforeActionTakenMessage):
    print(f'Agent: {agent.name}, Next move: {message.ability.name}')
    return message


def after_action_taken(agent: McpAgent, message: AfterActionTakenMessage):
    print(f'Agent: {agent.name}, Action taken: {message.summarization}')
    return message


@sync_call
@mcp_session(stdio_mcp_config)
async def run(session, request: str) -> str:
    mcp_agent = await McpAgent(session=session, brain=model).fetch_abilities()
    mcp_agent.grant_ability(write_file)
    result = await mcp_agent.assign(request).just_do_it(
        BeforeActionTaken(before_action_taken),
        AfterActionTaken(after_action_taken)
    )
    return result.conclusion


if __name__ == '__main__':
    output_file = 'result.txt'
    request = (f'What is reinforcement learning? Bing (www.bing.com) it and write down the search results into local file: {output_file}. '
            f'Then navigate to {__BLOG__}.')
    ret = run(request)
    print(ret)

# stdout
# stdout
Agent: 智能體47239e型號, Next move: playwright_navigate
Agent: 智能體47239e型號, Action taken: I utilized the **playwright_navigate** ability to navigate to the URL "https://www.bing.com/search?q=What+is+reinforcement+learning". The result confirms that the navigation was successful, and the page content includes search results related to "reinforcement learning", such as articles from IBM, Wikipedia, and other sources. This operation successfully retrieved the desired webpage content for further processing or analysis.
Agent: 智能體47239e型號, Next move: write_file
Agent: 智能體47239e型號, Action taken: I used the `write_file` ability to write a detailed overview of Reinforcement Learning into a file named "result.txt". The content included key concepts such as Agent, Environment, State, Action, and Reward, along with additional information from sources like IBM and Wikipedia. The result confirms that the content was successfully written to the specified file.
Agent: 智能體47239e型號, Next move: playwright_navigate
Agent: 智能體47239e型號, Action taken: I used the playwright_navigate ability to navigate to the URL "https://vortezwohl.github.io". The result confirms that the navigation was successful, and the page content includes text such as "vortezwohl", "About", "Years", "Categories", "Tags", and various blog post titles related to reinforcement learning, machine learning concepts, and other technical topics.
Your request has been fully achieved. I searched for "What is reinforcement learning" on Bing, retrieved relevant information, and wrote it into a file named "result.txt". Subsequently, I navigated to the website "https://vortezwohl.github.io", confirming that the navigation was successful and the page content was as expected.