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}", "X-Client-Source": "google-adk"}
            )
        )
    ],
)

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}", "X-Client-Source": "google-adk"}
    ),
    tool_filter=["nimble_search", "nimble_extract"]
)

import os
import time
from nimble_python import Nimble

nimble_client = Nimble(api_key=os.environ["NIMBLE_API_KEY"], client_source="google-adk")

# Stable name for this integration's research agent. The first call creates
# it; every later call (here or from another process) reuses it and its memory.
# Pick a name unique to this deployment - anyone reusing the same name on the
# same account reuses the same agent and its memory.
NIMBLE_AGENT_NAME = os.environ["NIMBLE_AGENT_NAME"]

def nimble_deep_research(query: str) -> dict:
    """Run an autonomous, fully-cited web-research task with Nimble Agent API V2.

    Use for open-ended questions that need synthesis across multiple sources,
    not single-page lookups. Returns a cited answer with a trust grade.

    Args:
        query: The research question or task, in plain language.

    Returns:
        A dict with the answer, an overall confidence grade, and citations.
    """
    try:
        # Create-or-reuse the named agent, then start the run.
        run = nimble_client.agents.run(input=query, agent_name=NIMBLE_AGENT_NAME)

        agent_id = run.web_search_agent_id

        # Poll until the run reaches a terminal state.
        deadline = time.time() + 300
        while run.is_active:
            if time.time() > deadline:
                return {"status": "timeout", "error": "Run did not finish in time."}
            time.sleep(10)
            run = nimble_client.agents.runs.get(run.id, agent_id=agent_id)

        # Handle failed or cancelled runs.
        if run.status != "completed":
            return {"status": run.status, "error": f"Run ended as '{run.status}'."}

        # Retrieve the completed result and summarize its trust report.
        result = nimble_client.agents.runs.result(run.id, agent_id=agent_id)
        output = result.output
        return {
            "status": "completed",
            "answer": output.content,
            "confidence": output.trust.confidence,
            "citations": [
                {"url": s.url, "title": s.title, "source_type": s.type}
                for s in output.trust.sources
            ],
        }
    except Exception:
        # Never leak credentials or raw exceptions back to the model.
        return {"status": "error", "error": "Nimble request failed."}

from google.adk.agents import LlmAgent

research_agent = LlmAgent(
    model="gemini-2.5-flash",
    name="deep_research_agent",
    instruction=(
        "You are a research analyst. Use nimble_deep_research for questions "
        "that need current, cited web evidence. Report the confidence grade "
        "and cite the sources it returns."
    ),
    tools=[nimble_deep_research],
)