构建一个MCP系统

import arxiv
import json
import os
from typing import List
from dotenv import load_dotenv
 
PAPER_DIR = "papers"
 
def search_papers(topic: str, max_results: int = 10) -> List[str]:
    """
    Search for papers on arXiv based on a given topic and store their information in a JSON file.
 
    Args:
        topic (str): The topic to search for.
        max_results (int, optional): The maximum number of results to return. Defaults to 10.
 
    Returns:
        List of paper IDs found in the search results.
    """
 
    client = arxiv.Client()
 
    search = arxiv.Search(
        query=topic, 
        max_results=max_results,
        sort_by=arxiv.SortCriterion.Relevance
    )
 
    papers = client.results(search)
 
    path = os.path.join(PAPER_DIR, topic.lower().replace(" ", "_"))
    os.makedirs(path, exist_ok=True)
 
    file_path = os.path.join(path, "papers_info.json")
 
    try:
        with open(file_path, "r") as json_file:
            papers_info = json.load(json_file)
    except (FileNotFoundError, json.JSONDecodeError):
        papers_info = {}
 
    paper_ids = []
 
    for paper in papers:
        paper_ids.append(paper.get_short_id())
        paper_info = {
            'title': paper.title,
            'authors': [author.name for author in paper.authors],
            'summary': paper.summary,
            'pdf_url': paper.pdf_url,
            'published': str(paper.published.date()),
        }
        papers_info[paper.get_short_id()] = paper_info
 
    with open(file_path, "w") as json_file:
        json.dump(papers_info, json_file, indent=2)
 
    print(f"Results are stored in: {file_path}")
 
    return paper_ids
 
def extract_info(paper_id: str) -> str:
    """
    Search for information about a specific paper across all topic directories.
 
    Args:
        paper_id (str): The ID of the paper to search for.
 
    Returns:
        JSON string with paper information if found, None otherwise.
    """
 
    for item in os.listdir(PAPER_DIR):
        item_path = os.path.join(PAPER_DIR, item)
        if os.path.isdir(item_path):
            file_path = os.path.join(item_path, "papers_info.json")
            if os.path.isfile(file_path):
                try:
                    with open(file_path, "r") as json_file:
                        papers_info = json.load(json_file)
                        if paper_id in papers_info:
                            return json.dumps(papers_info[paper_id], indent=2)
                except (FileNotFoundError, json.JSONDecodeError) as e:
                    print(f"Error reading {file_path}: {str(e)}")
                    continue
    
    return f"There is no saved information about {paper_id}"
 

mapping_tool_funtions = {
    "search_papers": search_papers,
    "extract_info": extract_info
}
 
def exectue_tool(tool_name: str, tool_args: dict):
 
    result = mapping_tool_funtions[tool_name](**tool_args)
 
    if result is None:
        result = "The operation completed but didn't return any results"
 
    elif isinstance(result, list):
        result = ','.join(result)
 
    elif isinstance(result, dict):
        result = json.dumps(result, indent=2)
 
    else:
        result = str(result)
 
    return result

import os
import json
from dotenv import load_dotenv
from openai import OpenAI
 
load_dotenv()
 
MODEL_PREFIX = "GEMINI"
 
api_key = os.getenv(f"{MODEL_PREFIX}_API_KEY")
base_url = os.getenv(f"{MODEL_PREFIX}_BASE_URL")
 
client = OpenAI(api_key=api_key, base_url=base_url)
 
model = "gemini-2.0-flash-exp"
 
tools_function = [
    {
        "type": "function",
        "function": {
            "name": "search_papers",
            "description": "Search for papers on arXiv based on a given topic and store their information in a JSON file.",
            "parameters": {
                "type": "object",
                "properties": {
                    "topic": {
                        "type": "string",
                        "description": "The topic to search for.",
                    },
                    "max_results": {
                        "type": "integer",
                        "description": "The maximum number of results to return. Defaults to 10.",
                        "default": 5,
                    },
                },
                "required": ["topic"],
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "extract_info",
            "description": "Extract information from a paper based on its ID.",
            "parameters": {
                "type": "object",
                "properties": {
                    "paper_id": {
                        "type": "string",
                        "description": "The ID of the paper to extract information from.",
                    },
                },
                "required": ["paper_id"],
            }
        }
    }
]
 
PAPER_DIR = "papers"
 
def search_papers(topic: str, max_results: int = 10) -> List[str]:
    """
    Search for papers on arXiv based on a given topic and store their information in a JSON file.
 
    Args:
        topic (str): The topic to search for.
        max_results (int, optional): The maximum number of results to return. Defaults to 10.
 
    Returns:
        List of paper IDs found in the search results.
    """
 
    client = arxiv.Client()
 
    search = arxiv.Search(
        query=topic, 
        max_results=max_results,
        sort_by=arxiv.SortCriterion.Relevance
    )
 
    papers = client.results(search)
 
    path = os.path.join(PAPER_DIR, topic.lower().replace(" ", "_"))
    os.makedirs(path, exist_ok=True)
 
    file_path = os.path.join(path, "papers_info.json")
 
    try:
        with open(file_path, "r") as json_file:
            papers_info = json.load(json_file)
    except (FileNotFoundError, json.JSONDecodeError):
        papers_info = {}
 
    paper_ids = []
 
    for paper in papers:
        paper_ids.append(paper.get_short_id())
        paper_info = {
            'title': paper.title,
            'authors': [author.name for author in paper.authors],
            'summary': paper.summary,
            'pdf_url': paper.pdf_url,
            'published': str(paper.published.date()),
        }
        papers_info[paper.get_short_id()] = paper_info
 
    with open(file_path, "w") as json_file:
        json.dump(papers_info, json_file, indent=2)
 
    print(f"Results are stored in: {file_path}")
 
    return paper_ids
 
def extract_info(paper_id: str) -> str:
    """
    Search for information about a specific paper across all topic directories.
 
    Args:
        paper_id (str): The ID of the paper to search for.
 
    Returns:
        JSON string with paper information if found, None otherwise.
    """
 
    for item in os.listdir(PAPER_DIR):
        item_path = os.path.join(PAPER_DIR, item)
        if os.path.isdir(item_path):
            file_path = os.path.join(item_path, "papers_info.json")
            if os.path.isfile(file_path):
                try:
                    with open(file_path, "r") as json_file:
                        papers_info = json.load(json_file)
                        if paper_id in papers_info:
                            return json.dumps(papers_info[paper_id], indent=2)
                except (FileNotFoundError, json.JSONDecodeError) as e:
                    print(f"Error reading {file_path}: {str(e)}")
                    continue
    
    return f"There is no saved information about {paper_id}"
 
mapping_tool_funtions = {
    "search_papers": search_papers,
    "extract_info": extract_info
}
 
def exectue_tool(tool_name: str, tool_args: dict):
 
    result = mapping_tool_funtions[tool_name](**tool_args)
 
    if result is None:
        result = "The operation completed but didn't return any results"
 
    elif isinstance(result, list):
        result = ','.join(result)
 
    elif isinstance(result, dict):
        result = json.dumps(result, indent=2)
 
    else:
        result = str(result)
 
    return result
 
def process_query(query):
 
    messages = [{'role': 'user', 'content': query}]
 
    response = client.chat.completions.create(
        model=model,
        tools=tools_function,
        messages=messages,
        tool_choice="auto",
    ) 
 
    process_quert = True
 
    while process_quert:
        assistant_content = []
 
        for choice in response.choices:
            if choice.finish_reason == 'stop':
 
                print(choice.message.content)
                assistant_content.append(choice.message.content)
 
                if len(response.choices) == 1:
                    process_quert = False
 
            elif choice.finish_reason == 'tool_calls':
 
                assistant_content.append(choice)
                messages.append(choice.message)
 
                call_message = []
 
                for call in choice.message.tool_calls:
 
                    tool_args = call.function.arguments
                    tool_name = call.function.name
 
                    print(f"Calling tool: {tool_name} with args: {tool_args}")
 
                    result = exectue_tool(tool_name, json.loads(tool_args))
 
                    call_message.append({
                        "role": "function",
                        "name": tool_name,
                        "content": result
                    })
 
                messages.append(call_message)
 
                response = client.chat.completions.create(
                    model=model,
                    tools=tools_function,
                    messages=messages,
                    tool_choice="auto",
                ) 
 
                if len(response.choices) == 1 and response.choices[0].finish_reason == 'stop':
                    print(response.choices[0].message.content)
                    process_quert = False
 
def chat_loop():
    print("Type your queries or type 'quit' to exit")
    while True:
        try:
            query = input("\nQuery:").strip()
            if query.lower() == "quit":
                break
 
            process_query(query)
            print("\n")
        except Exception as e:
            print(f"An error occurred: {str(e)}")
 
 
if __name__ == "__main__":
    chat_loop()
 

import arxiv
import json
import os
from typing import List
from mcp.server.fastmcp import FastMCP
 
mcp = FastMCP("research")
 
PAPER_DIR = "papers"
 
@mcp.tool()
def search_papers(topic: str, max_results: int = 5) -> List[str]:
    """
    Search for papers on arXiv based on a given topic and store their information in a JSON file.
 
    Args:
        topic (str): The topic to search for.
        max_results (int, optional): The maximum number of results to return. Defaults to 5.
 
    Returns:
        List of paper IDs found in the search results.
    """
 
    client = arxiv.Client()
 
    search = arxiv.Search(
        query=topic, 
        max_results=max_results,
        sort_by=arxiv.SortCriterion.Relevance
    )
 
    papers = client.results(search)
 
    path = os.path.join(PAPER_DIR, topic.lower().replace(" ", "_"))
    os.makedirs(path, exist_ok=True)
 
    file_path = os.path.join(path, "papers_info.json")
 
    try:
        with open(file_path, "r") as json_file:
            papers_info = json.load(json_file)
    except (FileNotFoundError, json.JSONDecodeError):
        papers_info = {}
 
    paper_ids = []
 
    for paper in papers:
        paper_ids.append(paper.get_short_id())
        paper_info = {
            'title': paper.title,
            'authors': [author.name for author in paper.authors],
            'summary': paper.summary,
            'pdf_url': paper.pdf_url,
            'published': str(paper.published.date()),
        }
        papers_info[paper.get_short_id()] = paper_info
 
    with open(file_path, "w") as json_file:
        json.dump(papers_info, json_file, indent=2)
 
    print(f"Results are stored in: {file_path}")
 
    return paper_ids
 
@mcp.tool()
def extract_info(paper_id: str) -> str:
    """
    Search for information about a specific paper across all topic directories.
 
    Args:
        paper_id (str): The ID of the paper to search for.
 
    Returns:
        JSON string with paper information if found, None otherwise.
    """
 
    for item in os.listdir(PAPER_DIR):
        item_path = os.path.join(PAPER_DIR, item)
        if os.path.isdir(item_path):
            file_path = os.path.join(item_path, "papers_info.json")
            if os.path.isfile(file_path):
                try:
                    with open(file_path, "r") as json_file:
                        papers_info = json.load(json_file)
                        if paper_id in papers_info:
                            return json.dumps(papers_info[paper_id], indent=2)
                except (FileNotFoundError, json.JSONDecodeError) as e:
                    print(f"Error reading {file_path}: {str(e)}")
                    continue
    
    return f"There is no saved information about {paper_id}"
 
 
if __name__ == "__main__":
    mcp.run(transport="stdio")

npx @modelcontextprotocol/inspector uv run step4.py

import os
import json
import asyncio
import nest_asyncio
from dotenv import load_dotenv
from openai import OpenAI
from mcp import ClientSession, StdioServerParameters, types
from mcp.client.stdio import stdio_client
from typing import List
 
nest_asyncio.apply()
 
load_dotenv()
 
MODEL_PREFIX = "GEMINI"
 
api_key = os.getenv(f"{MODEL_PREFIX}_API_KEY")
base_url = os.getenv(f"{MODEL_PREFIX}_BASE_URL")
 
client = OpenAI(api_key=api_key, base_url=base_url)
 
model = "gemini-2.5-flash-preview-04-17"
 
class MCP_ChatBot:
 
    def __init__(self):
        self.session: ClientSession = None
        self.client = client
        self.available_tools: List[dict] = []
 
    async def process_query(self, query):
 
        messages = [{'role': 'user', 'content': query}]
 
        response = self.client.chat.completions.create(
            model=model,
            tools=self.available_tools,
            messages=messages,
            tool_choice="auto",
        ) 
 
        process_quert = True
 
        while process_quert:
            assistant_content = []
 
            for choice in response.choices:
                if choice.finish_reason == 'stop':
 
                    print(choice.message.content)
                    assistant_content.append(choice.message.content)
 
                    if len(response.choices) == 1:
                        process_quert = False
 
                elif choice.finish_reason == 'tool_calls':
 
                    assistant_content.append(choice)
                    messages.append(choice.message)
 
                    call_message = []
 
                    for call in choice.message.tool_calls:
 
                        tool_args = call.function.arguments
                        tool_name = call.function.name
 
                        print(f"Calling tool: {tool_name} with args: {tool_args}")
 
                        result = await self.session.call_tool(tool_name, arguments=json.loads(tool_args))
 
                        call_message.append({
                            "role": "function",
                            "name": tool_name,
                            "content": result.content[0].text
                        })
 
                    messages.append(call_message)
 
                    response = self.client.chat.completions.create(
                        model=model,
                        tools=self.available_tools,
                        messages=messages,
                        tool_choice="auto",
                    ) 
 
                    if len(response.choices) == 1 and response.choices[0].finish_reason == 'stop':
                        print(response.choices[0].message.content)
                        process_quert = False
 
    async def chat_loop(self):
        print("Type your queries or type 'quit' to exit")
        while True:
            try:
                query = input("\nQuery:").strip()
                if query.lower() == "quit":
                    break
 
                await self.process_query(query)
                print("\n")
            except Exception as e:
                print(f"An error occurred: {str(e)}")
 
    async def connect_to_server_and_run(self):
        server_params = StdioServerParameters(
            command="uv",
            args=["run", "step4.py"],
            env=None
        )
        async with stdio_client(server_params) as (read, write):
            async with ClientSession(read, write) as session:
                self.session = session
                await session.initialize()
 
                response = await session.list_tools()
 
                tools = response.tools
                print("\nConnecting to server with tools: ", [tool.name for tool in tools])
 
                self.available_tools = [{
                    "type": "function",
                    "function": {
                        "name": tool.name,
                        "description": tool.description,
                        "parameters": tool.inputSchema
                    }
                } for tool in tools]
 
                await self.chat_loop()
 
async def main():
    chatbot = MCP_ChatBot()
    await chatbot.connect_to_server_and_run()
 
if __name__ == "__main__":
    asyncio.run(main())
 

{
  "mcpServers": {
    "fetch": {
      "command": "uvx",
      "args": [
        "mcp-server-fetch"
      ]
    },
    "research": {
      "command": "uv",
      "args": [
        "run",
        "step4.py"
      ]
    },
    "filesystem": {
        "command": "npx",
        "args": [
            "-y",
            "@modelcontextprotocol/server-filesystem",
            "."
        ]
    }
  }
}

import os
import json
import asyncio
import nest_asyncio
from dotenv import load_dotenv
from openai import OpenAI
from mcp import ClientSession, StdioServerParameters, types
from mcp.client.stdio import stdio_client
from typing import List, Dict, TypedDict
from contextlib import AsyncExitStack
 
nest_asyncio.apply()
 
load_dotenv()
 
MODEL_PREFIX = "GEMINI"
 
api_key = os.getenv(f"{MODEL_PREFIX}_API_KEY")
base_url = os.getenv(f"{MODEL_PREFIX}_BASE_URL")
 
client = OpenAI(api_key=api_key, base_url=base_url)
 
model = "gemini-2.5-flash-preview-04-17"
 
class ToolFunction(TypedDict):
    name: str
    description: str
    parameters: dict
 
class ToolDefinition(TypedDict):
    type: str
    function: ToolFunction
 
class MCP_ChatBot:
 
    def __init__(self):
        self.sessions: List[ClientSession] = []
        self.exit_stack = AsyncExitStack()
        self.client = client
        self.available_tools: List[ToolDefinition] = []
        self.tool_to_session: Dict[str, ClientSession] = {}
 
    async def connect_to_server(self, server_name: str, server_config: dict) -> None:
        """Connet to a single MCP server."""
        try:
            server_params = StdioServerParameters(**server_config)
            stdio_transport = await self.exit_stack.enter_async_context(
                stdio_client(server_params)
            )
            read, write = stdio_transport
            session = await self.exit_stack.enter_async_context(
                ClientSession(read, write)
            )
            await session.initialize()
            self.sessions.append(session)
 
            response = await session.list_tools()
 
            tools = response.tools
            print("\nConnecting to server with tools: ", [tool.name for tool in tools])
 
            for tool in tools:
                self.tool_to_session[tool.name] = session
                self.available_tools.append({
                    "type": "function",
                    "function": {
                        "name": tool.name,
                        "description": tool.description,
                        "parameters": tool.inputSchema
                    }
                })
            
        except Exception as e:
            print(f"Faild to connet to ${server_name}: {e}")
 
    
    async def connect_to_servers(self):
        """Connect to all configured MCP Servers"""
        try:
            with open("mcp.json") as file:
                data = json.load(file)
            
            servers = data.get("mcpServers", {})
            for server_name, server_config in servers.items():
                await self.connect_to_server(server_name, server_config)
        except Exception as e:
            print(f"Error loading server configuration: {e}")
            raise
 
    async def process_query(self, query):
 
        messages = [{'role': 'user', 'content': query}]
 
        response = self.client.chat.completions.create(
            model=model,
            tools=self.available_tools,
            messages=messages,
            tool_choice="auto",
        ) 
 
        process_quert = True
 
        while process_quert:
            assistant_content = []
 
            for choice in response.choices:
                if choice.finish_reason == 'stop':
 
                    print(choice.message.content)
                    assistant_content.append(choice.message.content)
 
                    if len(response.choices) == 1:
                        process_quert = False
 
                elif choice.finish_reason == 'tool_calls':
 
                    assistant_content.append(choice)
                    messages.append(choice.message)
 
                    call_message = []
 
                    for call in choice.message.tool_calls:
 
                        tool_args = call.function.arguments
                        tool_name = call.function.name
 
                        print(f"Calling tool: {tool_name} with args: {tool_args}")
 
                        session = self.tool_to_session[tool_name]
                        result = await session.call_tool(tool_name, arguments=json.loads(tool_args))
 
                        call_message.append({
                            "role": "function",
                            "name": tool_name,
                            "content": result.content[0].text
                        })
 
                    messages.append(call_message)
 
                    response = self.client.chat.completions.create(
                        model=model,
                        tools=self.available_tools,
                        messages=messages,
                        tool_choice="auto",
                    ) 
 
                    if len(response.choices) == 1 and response.choices[0].finish_reason == 'stop':
                        print(response.choices[0].message.content)
                        process_quert = False
 
    async def chat_loop(self):
        print("Type your queries or type 'quit' to exit")
        while True:
            try:
                query = input("\nQuery:").strip()
                if query.lower() == "quit":
                    break
 
                await self.process_query(query)
                print("\n")
            except Exception as e:
                print(f"An error occurred: {str(e)}")
 
    async def cleanup(self):
        await self.exit_stack.aclose()
 
async def main():
    chatbot = MCP_ChatBot()
    try:
        await chatbot.connect_to_servers()
        await chatbot.chat_loop()
    finally:
        await chatbot.cleanup()
 
if __name__ == "__main__":
    asyncio.run(main())
 

uv run step6.py