import os
import asyncio
from pydantic_ai importAgent
from pydantic_ai.mcp importMCPServerStdio

from mcp.server.fastmcp importFastMCP
from dotenv import load_dotenv
import os
from exa_py importExa

from mcp.server.fastmcp importFastMCP
import io
import base64
import matplotlib.pyplot as plt
import sys
from io importStringIO
import traceback

from mcp.server.fastmcp importFastMCP
import io
import base64
import matplotlib.pyplot as plt
import sys
from io importStringIO
import traceback


mcp =FastMCP("python_tools")


classPythonREPL:
def run(self, code):
        old_stdout = sys.stdout
        redirected_output = sys.stdout =StringIO()


try:
exec(code, globals())
            sys.stdout = old_stdout
return redirected_output.getvalue()
exceptExceptionas e:
            sys.stdout = old_stdout
return f"錯誤：{str(e)}\n{traceback.format_exc()}"


repl =PythonREPL()


@mcp.tool()
async def python_repl(code: str)-> str:
"""執(zhí)行 Python 代碼。"""
return repl.run(code)


@mcp.tool()
async def data_visualization(code: str)-> str:
"""執(zhí)行 Python 代碼，使用 Matplotlib 進(jìn)行可視化。"""
try:
        repl.run(code)
        buf = io.BytesIO()
        plt.savefig(buf, format='png')
        buf.seek(0)
        img_str = base64.b64encode(buf.getvalue()).decode()
        plt.close()# 關(guān)閉圖像以釋放內(nèi)存
return f"data:image/png;base64,{img_str}"
exceptExceptionas e:
return f"創(chuàng)建圖表時出錯：{str(e)}"


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

from mcp.server.fastmcp importFastMCP
from dotenv import load_dotenv
import os
from exa_py importExa


load_dotenv(override=True)


# 初始化 FastMCP
mcp =FastMCP(
    name="websearch",
    versinotallow="1.0.0",
    descriptinotallow="使用 Exa API 的網(wǎng)頁搜索功能"
)


# 初始化 Exa 客戶端
exa_api_key = os.getenv("EXA_API_KEY","")
exa =Exa(api_key=exa_api_key)


# 默認(rèn)搜索配置
websearch_config ={
"parameters":{
"default_num_results":5,
"include_domains":[]
}
}


@mcp.tool()
async def search_web(query: str, num_results:int=None)-> str:
"""使用 Exa API 搜索網(wǎng)頁并以 Markdown 格式返回結(jié)果。"""
try:
        search_args ={
"num_results": num_results or websearch_config["parameters"]["default_num_results"]
}


        search_results = exa.search_and_contents(
            query,
            summary={"query":"主要內(nèi)容和關(guān)鍵要點(diǎn)"},
**search_args
)


return format_search_results(search_results)
exceptExceptionas e:
return f"使用 Exa 搜索時發(fā)生錯誤：{e}"


def format_search_results(search_results):
ifnot search_results.results:
return"未找到結(jié)果。"


    markdown_results ="### 搜索結(jié)果：\n\n"
for idx, result in enumerate(search_results.results,1):
        title = result.title if hasattr(result,'title')and result.title else"無標(biāo)題"
        url = result.url
        published_date = f"（發(fā)布日期：{result.published_date}）"if hasattr(result,'published_date')and result.published_date else""


        markdown_results += f"**{idx}.** [{title}]({url}){published_date}\n"


if hasattr(result,'summary')and result.summary:
            markdown_results += f"> **摘要：** {result.summary}\n\n"
else:
            markdown_results +="\n"


return markdown_results


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

import os
import asyncio
from pydantic_ai importAgent
from pydantic_ai.mcp importMCPServerStdio
from pydantic_ai.models.openai importOpenAIMode


deepseek_chat_model =OpenAIModel(# 定義基礎(chǔ)為 Open AI
'deepseek-chat',
    base_url='https://api.deepseek.com',
    api_key=os.environ["DEEPSEEK_API_KEY"],
)


# 定義 MCP 服務(wù)器
exa_server =MCPServerStdio(
'python',
['exa_search.py']
)


python_tools_server =MCPServerStdio(
'python',
['python_tools.py']
)


# 定義包含兩個 MCP 服務(wù)器的代理
agent =Agent(
    deepseek_chat_model,
    mcp_servers=[exa_server, python_tools_server],
    retries=3
)


# 主異步函數(shù)
async def main():
    async with agent.run_mcp_servers():
        result = await agent.run("""
我需要分析一些氣候數(shù)據(jù)。首先，搜索近期的氣候變化統(tǒng)計(jì)數(shù)據(jù)。
然后，創(chuàng)建一個柱狀圖，展示過去十年全球氣溫的上升情況。
使用Python進(jìn)行數(shù)據(jù)可視化。
""")
print(result)


# 運(yùn)行異步函數(shù)
if __name__ =="__main__":
    asyncio.run(main())