Drop live, verifiable data into your agent framework.

pip install langchain-mcp-adapters langgraph "langchain[anthropic]"

import asyncio
from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.prebuilt import create_react_agent

async def main():
    # Keyless remote MCP server over streamable HTTP, no auth, no local process.
    client = MultiServerMCPClient({
        "dynamicfeed": {
            "transport": "streamable_http",      # "http" / "streamable-http" also accepted
            "url": "https://dynamicfeed.ai/mcp",
        }
    })

    tools = await client.get_tools()             # full public catalogue
    agent = create_react_agent("anthropic:claude-sonnet-4-6", tools)

    resp = await agent.ainvoke(
        {"messages": [{"role": "user", "content": "What is the weather in Sydney?"}]}
    )
    print(resp["messages"][-1].content)

if __name__ == "__main__":
    asyncio.run(main())

pip install llama-index-tools-mcp llama-index-llms-openai llama-index-core

import asyncio
from llama_index.tools.mcp import BasicMCPClient, McpToolSpec
from llama_index.core.agent.workflow import FunctionAgent
from llama_index.llms.openai import OpenAI

async def main():
    # A URL ending in /mcp auto-selects Streamable HTTP. Keyless.
    client = BasicMCPClient("https://dynamicfeed.ai/mcp")
    tools = await McpToolSpec(client=client).to_tool_list_async()   # full catalogue

    agent = FunctionAgent(
        tools=tools,
        llm=OpenAI(model="gpt-4.1"),             # set OPENAI_API_KEY
        system_prompt="You have live, verifiable data tools.",
    )
    print(str(await agent.run("What is the weather in Sydney?")))

if __name__ == "__main__":
    asyncio.run(main())

pip install 'crewai-tools[mcp]' crewai

from crewai import Agent, Task, Crew, Process
from crewai_tools import MCPServerAdapter

# "transport" is a KEY inside the dict (not a kwarg). Keyless: no headers.
server_params = {"url": "https://dynamicfeed.ai/mcp", "transport": "streamable-http"}

with MCPServerAdapter(server_params, connect_timeout=60) as mcp_tools:
    agent = Agent(
        role="Live-data analyst",
        goal="Answer using fresh, verifiable data from Dynamic Feed.",
        backstory="Wired to a remote MCP feed of real-time, sourced data.",
        tools=mcp_tools,                         # full catalogue (or use a collection URL)
        llm="anthropic/claude-opus-4-8",         # set ANTHROPIC_API_KEY
    )
    task = Task(
        description="Current weather in Sydney? Cite the source and timestamp.",
        expected_output="Conditions with source and measured_at.",
        agent=agent,
    )
    print(Crew(agents=[agent], tasks=[task], process=Process.sequential).kickoff())

pip install "pydantic-ai-slim[mcp]"

import asyncio
from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServerStreamableHTTP

# Keyless remote MCP server over Streamable HTTP.
server = MCPServerStreamableHTTP("https://dynamicfeed.ai/mcp")
agent = Agent("anthropic:claude-opus-4-8", toolsets=[server])

async def main():
    async with agent:                            # opens MCP connections once
        result = await agent.run("What is the weather in Sydney?")
        print(result.output)

if __name__ == "__main__":
    asyncio.run(main())

npm i ai @ai-sdk/mcp @ai-sdk/anthropic @modelcontextprotocol/sdk zod

import { createMCPClient } from '@ai-sdk/mcp';
import { StreamableHTTPClientTransport } from '@modelcontextprotocol/sdk/client/streamableHttp.js';
import { generateText, stepCountIs } from 'ai';
import { anthropic } from '@ai-sdk/anthropic';

// Keyless remote streamable-HTTP transport, no headers needed.
const transport = new StreamableHTTPClientTransport(new URL('https://dynamicfeed.ai/mcp'));
const mcp = await createMCPClient({ transport });

try {
  const tools = await mcp.tools();               // adapts the full public catalogue
  const { text } = await generateText({
    model: anthropic('claude-opus-4-8'),
    tools,
    stopWhen: stepCountIs(5),                     // allow multi-step tool calls
    prompt: 'Get the current AI model pricing and the latest CVEs.',
  });
  console.log(text);
} finally {
  await mcp.close();
}

pip install openai-agents

import asyncio
from agents import Agent, Runner
from agents.mcp import MCPServerStreamableHttp

async def main():
    # Keyless remote MCP over streamable HTTP, params needs only the URL.
    async with MCPServerStreamableHttp(
        name="Dynamic Feed",
        params={"url": "https://dynamicfeed.ai/mcp"},
        cache_tools_list=True,                   # cache the discovered catalogue
    ) as df:
        agent = Agent(
            name="Assistant",
            instructions="Answer with fresh, verifiable data from Dynamic Feed.",
            model="gpt-4.1",                     # set OPENAI_API_KEY
            mcp_servers=[df],
        )
        result = await Runner.run(agent, "What is the weather in Sydney?")
        print(result.final_output)

if __name__ == "__main__":
    asyncio.run(main())

pip install agent-framework --pre

import asyncio
from agent_framework import ChatAgent, MCPStreamableHTTPTool
from agent_framework.openai import OpenAIChatClient

async def main():
    # Keyless remote MCP, just name + url, no headers.
    async with (
        MCPStreamableHTTPTool(name="Dynamic Feed", url="https://dynamicfeed.ai/mcp") as df,
        ChatAgent(
            chat_client=OpenAIChatClient(model_id="gpt-4o-mini"),   # set OPENAI_API_KEY
            name="LiveDataAgent",
            instructions="Answer with fresh, sourced data from Dynamic Feed.",
            tools=df,
        ) as agent,
    ):
        result = await agent.run("What is the current AQI in Sydney?")
        print(result.text)

if __name__ == "__main__":
    asyncio.run(main())

pip install google-adk

import asyncio
from google.adk.agents import LlmAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.tools.mcp_tool import McpToolset
from google.adk.tools.mcp_tool.mcp_session_manager import StreamableHTTPConnectionParams
from google.genai import types

# Keyless remote MCP over streamable HTTP, no headers.
toolset = McpToolset(
    connection_params=StreamableHTTPConnectionParams(url="https://dynamicfeed.ai/mcp")
)
agent = LlmAgent(
    model="gemini-flash-latest",                 # set GOOGLE_API_KEY
    name="dynamicfeed_agent",
    instruction="Answer using the live Dynamic Feed tools.",
    tools=[toolset],                             # auto-discovers the public catalogue
)

async def main():
    sessions = InMemorySessionService()
    await sessions.create_session(app_name="df", user_id="u1", session_id="s1")
    runner = Runner(app_name="df", agent=agent, session_service=sessions)
    msg = types.Content(role="user", parts=[types.Part(text="Weather in Sydney?")])
    async for ev in runner.run_async(user_id="u1", session_id="s1", new_message=msg):
        if ev.is_final_response() and ev.content:
            print(ev.content.parts[0].text)

if __name__ == "__main__":
    asyncio.run(main())

npm install @mastra/mcp @mastra/core

import { MCPClient } from "@mastra/mcp";
import { Agent } from "@mastra/core/agent";

// Keyless remote MCP, a bare url tries Streamable HTTP first.
const mcp = new MCPClient({
  servers: { dynamicfeed: { url: new URL("https://dynamicfeed.ai/mcp") } },
});

const tools = await mcp.listTools();             // namespaced public catalogue

const agent = new Agent({
  name: "Dynamic Feed Agent",
  instructions: "Answer using live Dynamic Feed tools; cite source + timestamp.",
  model: "openai/gpt-4o",                        // set OPENAI_API_KEY
  tools,
});

const res = await agent.generate("Latest published OpenAI model and its price?");
console.log(res.text);
await mcp.disconnect();