Google ADK

import os
from google.adk.agents import LlmAgent
from google.adk.tools.mcp_tool import McpToolset
from google.adk.tools.mcp_tool.mcp_session_manager import StreamableHTTPConnectionParams

NIMBLE_API_KEY = os.environ["NIMBLE_API_KEY"]

root_agent = LlmAgent(
    model="gemini-2.5-flash",
    name="web_research_agent",
    instruction=(
        "You are a research assistant with access to real-time web data. "
        "Use the available tools to search the web, extract content from URLs, "
        "crawl sites, and discover URLs."
    ),
    tools=[
        McpToolset(
            connection_params=StreamableHTTPConnectionParams(
                url="https://mcp.nimbleway.com/mcp",
                headers={"Authorization": f"Bearer {NIMBLE_API_KEY}"}
            )
        )
    ],
)

import asyncio
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService

async def main():
    session_service = InMemorySessionService()
    session = await session_service.create_session(
        app_name="nimble_app", user_id="user"
    )

    runner = Runner(
        agent=root_agent,
        app_name="nimble_app",
        session_service=session_service,
    )

    from google.genai import types

    response = await runner.run_async(
        user_id="user",
        session_id=session.id,
        new_message=types.Content(
            role="user",
            parts=[types.Part(text="What are the latest trends in AI agents?")]
        ),
    )

    for event in response:
        if event.content and event.content.parts:
            for part in event.content.parts:
                if part.text:
                    print(part.text)

asyncio.run(main())

McpToolset(
    connection_params=StreamableHTTPConnectionParams(
        url="https://mcp.nimbleway.com/mcp",
        headers={"Authorization": f"Bearer {NIMBLE_API_KEY}"}
    ),
    tool_filter=["search", "extract"]
)