支持MCP的7大AI框架

from openai import OpenAI

client = OpenAI(
    base_url="https://api.opentools.com",
    api_key="<OPENTOOLS_API_KEY>"
)

completion = client.chat.completions.create(
    model="anthropic/claude-3.7-sonnet",
    messages=[
        { "role": "user", "content": "Compare specs of top 5 EVs on caranddriver.com" }
    ],
    tools=[{ "type": "mcp", "ref": "firecrawl" }]
)

import asyncio
import shutil
import streamlit as st
from agents import Agent, Runner, trace
from agents.mcp import MCPServer, MCPServerStdio

async def query_git_repo(mcp_server: MCPServer, directory_path: str, query: str):
    agent = Agent(
        name="Assistant",
        instructions=f"Answer questions about the localgit repository at {directory_path}, use that for repo_path",
        mcp_servers=[mcp_server],
    )

    with st.spinner(f"Running query: {query}"):
        result = await Runner.run(starting_agent=agent, input=query)
        return result.final_output

async def run_streamlit_app():
    st.title("Local Git Repo Explorer")
    st.write("This app allows you to query information about a local git repository.")

    directory_path = st.text_input("Enter the path to the git repository:")

    if directory_path:
        # Common queries as buttons
        col1, col2 = st.columns(2)
        with col1:
            if st.button("Most frequent contributor"):
                query = "Who's the most frequent contributor?"
                run_query(directory_path, query)

        with col2:
            if st.button("Last change summary"):
                query = "Summarize the last change in the repository."
                run_query(directory_path, query)

        # Custom query
        custom_query = st.text_input("Or enter your own query:")
        if st.button("Run Custom Query") and custom_query:
            run_query(directory_path, custom_query)

def run_query(directory_path, query):
    if not shutil.which("uvx"):
        st.error("uvx is not installed. Please install it with `pip install uvx`.")
        return

    async def execute_query():
        async with MCPServerStdio(
            cache_tools_list=True,
            params={
                "command": "python", 
                "args": [
                    "-m", 
                    "mcp_server_git", 
                    "--repository", 
                    directory_path
                ]
            },
        ) as server:
            with trace(workflow_name="MCP Git Query"):
                result = await query_git_repo(server, directory_path, query)
                st.markdown("### Result")
                st.write(result)

    asyncio.run(execute_query())

if __name__ == "__main__":
    st.set_page_config(
        page_title="Local Git Repo Explorer",
        page_icon="📊",
        layout="centered"
    )
    # Change from async to synchronous implementation
    # Since Streamlit doesn't work well with asyncio in the main thread

    # Define a synchronous version of our app
    def main_streamlit_app():
        st.title("Local Git Repo Explorer")
        st.write("This app allows you to query information about a Git repository.")

        directory_path = st.text_input("Enter the path to the git repository:")

        if directory_path:
            # Common queries as buttons
            col1, col2 = st.columns(2)
            with col1:
                if st.button("Most frequent contributor"):
                    query = "Who's the most frequent contributor?"
                    run_query(directory_path, query)

            with col2:
                if st.button("Last change summary"):
                    query = "Summarize the last change in the repository."
                    run_query(directory_path, query)

            # Custom query
            custom_query = st.text_input("Or enter your own query:")
            if st.button("Run Custom Query") and custom_query:
                run_query(directory_path, custom_query)

    # Run the synchronous app
    main_streamlit_app()

import streamlit as st
from praisonaiagents import Agent, MCP

st.title("🏠 Airbnb Booking Assistant")

# Create the agent
@st.cache_resource
def get_agent():
    return Agent(
        instructions="""You help book apartments on Airbnb.""",
        llm="gpt-4o-mini",
        tools=MCP("npx -y @openbnb/mcp-server-airbnb --ignore-robots-txt")
    )

# Initialize chat history
if "messages" not in st.session_state:
    st.session_state.messages = []

# Display chat history
for message in st.session_state.messages:
    with st.chat_message(message["role"]):
        st.markdown(message["content"])

# User input form
with st.form("booking_form"):
    st.subheader("Enter your booking details")

    destination = st.text_input("Destination:", "Paris")

    col1, col2 = st.columns(2)
    with col1:
        check_in = st.date_input("Check-in date")
    with col2:
        check_out = st.date_input("Check-out date")

    adults = st.number_input("Number of adults:", min_value=1, max_value=10, value=2)

    submitted = st.form_submit_button("Search for accommodations")

    if submitted:
        search_agent = get_agent()

        # Format the query
        query = f"I want to book an apartment in {destination} from {check_in.strftime('%m/%d/%Y')} to {check_out.strftime('%m/%d/%Y')} for {adults} adults"

        # Add user message to chat history
        st.session_state.messages.append({"role": "user", "content": query})

        # Display user message
        with st.chat_message("user"):
            st.markdown(query)

        # Get response from the agent
        with st.chat_message("assistant"):
            with st.spinner("Searching for accommodations..."):
                response = search_agent.start(query)
                st.markdown(response)

        # Add assistant response to chat history
        st.session_state.messages.append({"role": "assistant", "content": response})

# Allow for follow-up questions
if st.session_state.messages:
    prompt = st.chat_input("Ask a follow-up question about the accommodations")
    if prompt:
        search_agent = get_agent()

        # Add user message to chat history
        st.session_state.messages.append({"role": "user", "content": prompt})

        # Display user message
        with st.chat_message("user"):
            st.markdown(prompt)

        # Get response from the agent
        with st.chat_message("assistant"):
            with st.spinner("Thinking..."):
                response = search_agent.start(prompt)
                st.markdown(response)

        # Add assistant response to chat history
        st.session_state.messages.append({"role": "assistant", "content": response}) 

# Copyright (C) 2024 Andrew Wason
# SPDX-License-Identifier: MIT
import asyncio
import pathlib
import sys
import typing as t

from langchain_core.messages import AIMessage, BaseMessage, HumanMessage
from langchain_core.output_parsers import StrOutputParser
from langchain_core.tools import BaseTool
from langchain_groq import ChatGroq
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from langchain_mcp import MCPToolkit


async def run(tools: list[BaseTool], prompt: str) -> str:
    model = ChatGroq(model_name="llama-3.1-8b-instant", stop_sequences=None)  # requires GROQ_API_KEY
    tools_map = {tool.name: tool for tool in tools}
    tools_model = model.bind_tools(tools)
    messages: list[BaseMessage] = [HumanMessage(prompt)]
    ai_message = t.cast(AIMessage, await tools_model.ainvoke(messages))
    messages.append(ai_message)
    for tool_call in ai_message.tool_calls:
        selected_tool = tools_map[tool_call["name"].lower()]
        tool_msg = await selected_tool.ainvoke(tool_call)
        messages.append(tool_msg)
    return await (tools_model | StrOutputParser()).ainvoke(messages)


async def main(prompt: str) -> None:
    server_params = StdioServerParameters(
        command="npx",
        args=["-y", "@modelcontextprotocol/server-filesystem", str(pathlib.Path(__file__).parent.parent)],
    )
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            toolkit = MCPToolkit(session=session)
            await toolkit.initialize()
            response = await run(toolkit.get_tools(), prompt)
            print(response)


if __name__ == "__main__":
    prompt = sys.argv[1] if len(sys.argv) > 1 else "Read and summarize the file ./readme.md"
    asyncio.run(main(prompt))

# pip install chainlit

import chainlit as cl
from mcp import ClientSession

@cl.on_mcp_connect
async def on_mcp_connect(connection, session: ClientSession):
    """Called when an MCP connection is established"""
    # Your connection initialization code here
    # This handler is required for MCP to work

@cl.on_mcp_disconnect
async def on_mcp_disconnect(name: str, session: ClientSession):
    """Called when an MCP connection is terminated"""
    # Optional handler: Cleanup your code here

    # Define server parameters
    airbnb_server_params = StdioServerParameters(
        command="npx",
        args=["-y", "@openbnb/mcp-server-airbnb", "--ignore-robots-txt"],
        env=env,
    )

    maps_server_params = StdioServerParameters(
        command="npx", args=["-y", "@modelcontextprotocol/server-google-maps"], env=env
    )

    # Use contextlib.AsyncExitStack to manage multiple async context managers
    async with contextlib.AsyncExitStack() as stack:
        # Create stdio clients for each server
        airbnb_client, _ = await stack.enter_async_context(stdio_client(airbnb_server_params))
        maps_client, _ = await stack.enter_async_context(stdio_client(maps_server_params))

        # Create all agents
        airbnb_agent = Agent(
            name="Airbnb",
            role="Airbnb Agent",
            model=OpenAIChat("gpt-4o"),
            tools=[airbnb_client],
            instructions=dedent("""\
                You are an agent that can find Airbnb listings for a given location.\
            """),
            add_datetime_to_instructions=True,
        )

import os
from dotenv import load_dotenv
from upsonic import Task, Agent, Direct
from upsonic.client.tools import Search  # Adding Search as a fallback tool

# Load environment variables from .env file
load_dotenv()

# Get the OpenAI API key from environment variables
openai_api_key = os.getenv("OPENAI_API_KEY")
if not openai_api_key:
    raise ValueError("OPENAI_API_KEY not found in .env file")

# Set your OpenAI API key for the session
os.environ["OPENAI_API_KEY"] = openai_api_key

# Define the HackerNews MCP tool
# Using the correct MCP setup for HackerNews based on Upsonic documentation
class HackerNewsMCP:
    command = "uvx"
    args = ["mcp-hn"]
    # No environment variables are needed for this MCP

# Create a task to analyze the latest HackerNews stories
# Adding Search as a fallback in case HackerNews MCP fails
task = Task(
    "Analyze the top 5 HackerNews stories for today. Provide a brief summary of each story, "
    "identify any common themes or trends, and highlight which stories might be most relevant "
    "for someone interested in AI and software development.",
    tools=[HackerNewsMCP, Search]  # Include both HackerNews MCP and Search tools
)

# Create an agent specialized in tech news analysis
agent = Agent(
    "Tech News Analyst",
    company_url="https://news.ycombinator.com/",
    company_objective="To provide insightful analysis of tech industry news and trends"
)

# Execute the task with the agent and print the results
print("Analyzing HackerNews stories...")
agent.print_do(task)

# Alternatively, you can use a Direct LLM call if the task is straightforward
# print("Direct analysis of HackerNews stories...")
# Direct.print_do(task)

# If you want to access the response programmatically:
# agent.do(task)
# result = task.response
# print(result)