Pipedream Auth

Connect Agno to Pipedream with an authenticated app and user context.

This example selects Slack through x-pd-app-slug. Change that value to your integration's app slug. The external user ID selects the connected accounts for that user in your project.

Code

import asyncio
import os

from agno.agent import Agent
from agno.models.openai import OpenAIResponses
from agno.tools.mcp import MCPTools, StreamableHTTPClientParams
from agno.utils.log import log_exception

mcp_access_token = os.environ["MCP_ACCESS_TOKEN"]
pipedream_project_id = os.environ["PIPEDREAM_PROJECT_ID"]
pipedream_environment = os.environ["PIPEDREAM_ENVIRONMENT"]
pipedream_external_user_id = os.environ["PIPEDREAM_EXTERNAL_USER_ID"]

server_params = StreamableHTTPClientParams(
    url="https://remote.mcp.pipedream.net/v3",
    headers={
        "Authorization": f"Bearer {mcp_access_token}",
        "x-pd-project-id": pipedream_project_id,
        "x-pd-environment": pipedream_environment,
        "x-pd-external-user-id": pipedream_external_user_id,
        "x-pd-app-slug": "slack",
    },
)


async def run_agent(task: str) -> None:
    try:
        async with MCPTools(
            server_params=server_params,
            transport="streamable-http",
            timeout_seconds=20,
        ) as mcp:
            agent = Agent(
                model=OpenAIResponses(id="gpt-5.2"),
                tools=[mcp],
                markdown=True,
            )
            await agent.aprint_response(
                input=task,
                stream=True,
            )
    except Exception as e:
        log_exception(f"Unexpected error: {e}")


if __name__ == "__main__":
    asyncio.run(
        run_agent("Show me the latest message in the channel #general")
    )

Prerequisites

Create a Pipedream Connect project and obtain an MCP access token by following the Pipedream developer setup. Use a stable ID from your application for PIPEDREAM_EXTERNAL_USER_ID. If the first tool call returns a connection link, open it to authorize the Slack account, then run the example again.

uv pip install "agno[mcp]" openai
export MCP_ACCESS_TOKEN=***
export OPENAI_API_KEY=***
export PIPEDREAM_PROJECT_ID=proj_***
export PIPEDREAM_ENVIRONMENT=development
export PIPEDREAM_EXTERNAL_USER_ID=user_***

Run the Example

Save the code above as pipedream_auth.py, then run:

python pipedream_auth.py