AI agent 大模型实战之 智能体创建Vue项目

封装tools

import subprocess
from mcp.server.fastmcp import FastMCP
mcp = FastMCP()

def run_apple_script(script):
    p = subprocess.Popen(["osascript","-e",script],stdout=subprocess.PIPE,stderr=subprocess.PIPE)
    out, err = p.communicate()
    return out.decode("utf-8").strip(),err.decode("utf-8").strip()

@mcp.tool(name="close_terminal",description="关闭终端工具")
def close_terminal():
    output,error = run_apple_script("""
    tell application "Terminal" to quit
    """)

    if error:
        return error
    else:
        return True

@mcp.tool(name="open_terminal",description="打开终端工具")
def open_terminal():
    output,error = run_apple_script("""
        tell application "Terminal" to activate
    """)
    if error:
        return error
    else:
        return True


@mcp.tool(name="run_script_in_terminal",description="在终端中运行脚本命令")
def run_script_in_terminal(script):
    print(script)
    output,error = run_apple_script(f"""
        tell application "Terminal"
            activate
            if (count of windows) > 0 then
                do script "{script}" in window 1
            else
                do script "{script}"
            end if
        end tell
    """)
    if error:
        return error
    else:
        return output

@mcp.tool(name="get_terminal_full_text",description="查看终端返回的信息")
def get_terminal_full_text():
    output,error = run_apple_script(f"""
tell application "Terminal" 
    set fullText to history of selected tab of front window
end tell
""")
    if error:
        return error
    else:
        return output

 
if __name__=="__main__":
    mcp.run(transport="stdio")

from mcp_all.utils.tools import create_mcp_stdio_client


async def get_stdio_terminal_tools():
    params={
        "command":"python",
        "args":[
            "/Volumes/c/workspace/ai-agent-test/mcp_all/mcp/terminal_tools.py"
        ]
    }
    client,tools = await create_mcp_stdio_client("terminal_tools",params)
    return tools

调用

async def run_agent():
    #memory=   FileChatMessageHistory(f"{session_id}.json")
    memory = MemorySaver()
    tools  =  await get_stdio_terminal_tools()

    agent = create_agent(
        model=llm,
        tools=tools,
        checkpointer=memory,
        debug=False,
    )

    while True:
        user_input = input("用户:")

        if user_input.lower() == "exit" or user_input.lower() == "quit":
            break

        resp = await agent.ainvoke(
            input=({"messages": [{"role": "user", "content":user_input}]}),
            config={"configurable":{"thread_id":thread_id}}
        )
        print("助理:")
        print(resp)
        for chunk in resp:
            print(chunk,end="")
        print("\n")

运行，在命令行输入 在xxx目录 创建一个vue项目