MCP工具适配器

# If using pipenv (recommended)
pipenv install mcp langchain langchain-openai langgraph python-dotenv

# If using pip
pip install mcp langchain langchain-openai langgraph python-dotenv

OPENAI_API_KEY=your_actual_api_key_here

from dotenv import load_dotenv

# Load environment variables from .env file
load_dotenv()

import os
os.environ["OPENAI_API_KEY"] = "your_api_key_here"

import mcp
from mcp.server import expose

@expose()
def add(a: int, b: int) -> int:
    """Add two numbers and return the result."""
    return a + b

@expose()
def get_weather(city: str) -> str:
    """
    Get the current weather for a city.

    Args:
        city: The name of the city to get weather for
    """
    # In a real application, you'd call a weather API here
    return f"Weather in {city}: Sunny +11°C"

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

from mcp_langchain_adapter import MCPAdapter

# Create an adapter instance, pointing to your MCP server script
adapter = MCPAdapter("simple_server.py")

# Initialize the connection and get the list of available tools
tools = adapter.get_tools()

# Print the available tools
print(f"Found {len(tools)} tools:")
for tool in tools:
    print(f"- {tool.name}: {tool.description}")

from langchain.agents import AgentExecutor, create_react_agent
from langchain.prompts import PromptTemplate
from langchain_openai import ChatOpenAI

# Initialize the language model
llm = ChatOpenAI(model="gpt-3.5-turbo")

# Create a prompt template for the agent
template = """Answer the following questions as best you can using the provided tools.

Available tools:
{tools}

Use the following format:

Question: the input question you must answer
Thought: you should always think about what to do
Action: the action to take, should be one of [{tool_names}]
Action Input: the input to the action
Observation: the result of the action
... (this Thought/Action/Action Input/Observation can repeat N times)
Thought: I now know the final answer
Final Answer: the final answer to the original input question

Begin!

Question: {input}
Thought: """

prompt_template = PromptTemplate.from_template(template)

# Create a LangChain agent with the MCP tools
agent = create_react_agent(llm, tools, prompt_template)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# Run the agent
result = agent_executor.invoke({"input": "What is 5 + 7?"})
print(result["output"])

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
from langgraph.prebuilt import create_react_agent
from langgraph.checkpoint.memory import MemorySaver

# Initialize the language model
llm = ChatOpenAI(model="gpt-3.5-turbo")

# Create a memory saver for conversation history
memory = MemorySaver()

# Create a LangGraph react agent with the MCP tools
agent = create_react_agent(
    llm,
    tools,
    prompt="You are a helpful AI assistant that can use tools to solve problems.",
    checkpointer=memory
)

# Create the configuration with thread ID for memory
config = {"configurable": {"thread_id": "example-thread"}}

# Run the agent with a question
result = agent.invoke(
    {"messages": [HumanMessage(content="What is 5 + 7?")]},
    config
)

# Get the final answer
final_answer = result["messages"][-1].content
print(final_answer)

# Continue the conversation with a follow-up question
state = memory.get("example-thread")
messages = state["messages"] + [HumanMessage(content="What's the weather in London?")]
result = agent.invoke({"messages": messages}, config)
print(result["messages"][-1].content)

MCPAdapter(server_script_path: str, env: Dict[str, str] = None)

MCPToolWrapper(
    name: str,
    description: str,
    server_script_path: str,
    env: Optional[Dict[str, str]] = None,
    args_schema: Optional[Type[BaseModel]] = None
)

from mcp_langchain_adapter import MCPAdapter

# Create an adapter instance
adapter = MCPAdapter("simple_server.py")

# Get all tools
tools = adapter.get_tools()

# Print information about the tools
print(f"Found {len(tools)} tools:")
for tool in tools:
    print(f"- {tool.name}: {tool.description}")

# Use a specific tool
add_tool = adapter.get_tool_by_name("add")
if add_tool:
    result = add_tool.run({"a": 5, "b": 7})
    print(f"Result of add(5, 7): {result}")

# Use another tool
weather_tool = adapter.get_tool_by_name("get_weather")
if weather_tool:
    result = weather_tool.run({"city": "London"})
    print(f"Result of get_weather('London'): {result}")

import logging

logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler("mcp_server.log"),
        logging.StreamHandler()
    ]
)

# Rest of your MCP server code...