MCP (Model Context Protocol)

MCP integration for AI-Parrot.

MCPEnabledMixin

MCPEnabledMixin(*args, **kwargs)

Mixin to add complete MCP capabilities to agents.

Source code in packages/ai-parrot/src/parrot/mcp/integration.py

def __init__(self, *args, **kwargs):
    super().__init__(*args, **kwargs)
    self._mcp_initialized = True

add_mcp_server async

add_mcp_server(config: MCPClientConfig) -> List[str]

Add an MCP server with full feature support.

Source code in packages/ai-parrot/src/parrot/mcp/integration.py

async def add_mcp_server(self, config: MCPServerConfig) -> List[str]:
    """Add an MCP server with full feature support."""
    return await self.tool_manager.add_mcp_server(config)

add_local_mcp_server async

add_local_mcp_server(name: str, script_path: Union[str, Path], interpreter: str = 'python', **kwargs) -> List[str]

Add a local stdio MCP server.

Source code in packages/ai-parrot/src/parrot/mcp/integration.py

async def add_local_mcp_server(
    self,
    name: str,
    script_path: Union[str, Path],
    interpreter: str = "python",
    **kwargs
) -> List[str]:
    """Add a local stdio MCP server."""
    config = create_local_mcp_server(name, script_path, interpreter, **kwargs)
    return await self.add_mcp_server(config)

add_http_mcp_server async

add_http_mcp_server(name: str, url: str, auth_type: Optional[str] = None, auth_config: Optional[Dict[str, Any]] = None, headers: Optional[Dict[str, str]] = None, **kwargs) -> List[str]

Add an HTTP MCP server.

Source code in packages/ai-parrot/src/parrot/mcp/integration.py

async def add_http_mcp_server(
    self,
    name: str,
    url: str,
    auth_type: Optional[str] = None,
    auth_config: Optional[Dict[str, Any]] = None,
    headers: Optional[Dict[str, str]] = None,
    **kwargs
) -> List[str]:
    """Add an HTTP MCP server."""
    config = create_http_mcp_server(name, url, auth_type, auth_config, headers, **kwargs)
    return await self.add_mcp_server(config)

add_api_key_mcp_server async

add_api_key_mcp_server(name: str, url: str, api_key: str, header_name: str = 'X-API-Key', **kwargs) -> List[str]

Add an MCP server with API key auth.

Source code in packages/ai-parrot/src/parrot/mcp/integration.py

async def add_api_key_mcp_server(
    self,
    name: str,
    url: str,
    api_key: str,
    header_name: str = "X-API-Key",
    **kwargs
) -> List[str]:
    """Add an MCP server with API key auth."""
    config = create_api_key_mcp_server(name, url, api_key, header_name, **kwargs)
    return await self.add_mcp_server(config)

add_oauth_mcp_server async

add_oauth_mcp_server(name: str, url: str, user_id: str, oauth2: Optional[MCPOAuth2Config] = None, client_id: Optional[str] = None, auth_url: Optional[str] = None, token_url: Optional[str] = None, scopes: Optional[List[str]] = None, client_secret: Optional[str] = None, **kwargs) -> List[str]

Add an MCP server with OAuth2 authorization code support.

Accepts either a ready-built :class:~parrot.mcp.oauth2_config.MCPOAuth2Config (via the oauth2 parameter) or individual parameters for backward compatibility.

PARAMETER	DESCRIPTION
name	Server name / registry key. TYPE: str
url	MCP server base URL. TYPE: str
user_id	User identifier for token storage scoping. TYPE: str
oauth2	Pre-built OAuth2 configuration. TYPE: Optional[MCPOAuth2Config] DEFAULT: None
client_id	OAuth2 client ID (legacy). TYPE: Optional[str] DEFAULT: None
auth_url	Authorization endpoint (legacy). TYPE: Optional[str] DEFAULT: None
token_url	Token endpoint (legacy). TYPE: Optional[str] DEFAULT: None
scopes	OAuth2 scopes (legacy). TYPE: Optional[List[str]] DEFAULT: None
client_secret	OAuth2 client secret (legacy). TYPE: Optional[str] DEFAULT: None
**kwargs	Extra :class:~parrot.mcp.client.MCPClientConfig fields. DEFAULT: {}

RETURNS	DESCRIPTION
List[str]	List of registered MCP tool names.

Source code in packages/ai-parrot/src/parrot/mcp/integration.py

async def add_oauth_mcp_server(
    self,
    name: str,
    url: str,
    user_id: str,
    oauth2: Optional[MCPOAuth2Config] = None,
    # Legacy backward-compat params:
    client_id: Optional[str] = None,
    auth_url: Optional[str] = None,
    token_url: Optional[str] = None,
    scopes: Optional[List[str]] = None,
    client_secret: Optional[str] = None,
    **kwargs
) -> List[str]:
    """Add an MCP server with OAuth2 authorization code support.

    Accepts either a ready-built :class:`~parrot.mcp.oauth2_config.MCPOAuth2Config`
    (via the ``oauth2`` parameter) or individual parameters for backward
    compatibility.

    Args:
        name: Server name / registry key.
        url: MCP server base URL.
        user_id: User identifier for token storage scoping.
        oauth2: Pre-built OAuth2 configuration.
        client_id: OAuth2 client ID (legacy).
        auth_url: Authorization endpoint (legacy).
        token_url: Token endpoint (legacy).
        scopes: OAuth2 scopes (legacy).
        client_secret: OAuth2 client secret (legacy).
        **kwargs: Extra :class:`~parrot.mcp.client.MCPClientConfig` fields.

    Returns:
        List of registered MCP tool names.
    """
    config = create_oauth_mcp_server(
        name=name,
        url=url,
        user_id=user_id,
        oauth2=oauth2,
        client_id=client_id,
        client_secret=client_secret,
        auth_url=auth_url,
        token_url=token_url,
        scopes=scopes,
        **kwargs
    )
    return await self.add_mcp_server(config)

add_perplexity_mcp_server async

add_perplexity_mcp_server(api_key: str, name: str = 'perplexity', **kwargs) -> List[str]

Add a Perplexity MCP server capability.

Source code in packages/ai-parrot/src/parrot/mcp/integration.py

async def add_perplexity_mcp_server(
    self,
    api_key: str,
    name: str = "perplexity",
    **kwargs
) -> List[str]:
    """Add a Perplexity MCP server capability."""
    config = create_perplexity_mcp_server(api_key, name=name, **kwargs)
    return await self.add_mcp_server(config)

add_fireflies_mcp_server async

add_fireflies_mcp_server(api_key: Optional[str] = None, **kwargs) -> List[str]

Add Fireflies.ai MCP server capability.

The API key is resolved with the following precedence

1. Explicit api_key argument.
2. FIREFLIES_API_KEY environment variable (via navconfig.config).
3. ValueError if neither is available.

PARAMETER	DESCRIPTION
api_key	Fireflies API key (optional; falls back to FIREFLIES_API_KEY env var) TYPE: Optional[str] DEFAULT: None
**kwargs	Additional MCPServerConfig parameters DEFAULT: {}

RETURNS	DESCRIPTION
List[str]	List of registered tool names

Example

tools = await agent.add_fireflies_mcp_server( ... api_key="your-fireflies-api-key" ... )

Or rely on FIREFLIES_API_KEY env var:

tools = await agent.add_fireflies_mcp_server()

Source code in packages/ai-parrot/src/parrot/mcp/integration.py

async def add_fireflies_mcp_server(
    self,
    api_key: Optional[str] = None,
    **kwargs
) -> List[str]:
    """Add Fireflies.ai MCP server capability.

    The API key is resolved with the following precedence:
      1. Explicit ``api_key`` argument.
      2. ``FIREFLIES_API_KEY`` environment variable (via ``navconfig.config``).
      3. ``ValueError`` if neither is available.

    Args:
        api_key: Fireflies API key (optional; falls back to FIREFLIES_API_KEY env var)
        **kwargs: Additional MCPServerConfig parameters

    Returns:
        List of registered tool names

    Example:
        >>> tools = await agent.add_fireflies_mcp_server(
        ...     api_key="your-fireflies-api-key"
        ... )
        >>> # Or rely on FIREFLIES_API_KEY env var:
        >>> tools = await agent.add_fireflies_mcp_server()
    """
    config = create_fireflies_mcp_server(api_key=api_key, **kwargs)
    return await self.add_mcp_server(config)

add_chrome_devtools_mcp_server async

add_chrome_devtools_mcp_server(browser_url: str = 'http://127.0.0.1:9222', name: str = 'chrome-devtools', **kwargs) -> List[str]

Add Chrome DevTools MCP server capability.

PARAMETER	DESCRIPTION
browser_url	URL where Chrome is listening for devtools protocol TYPE: str DEFAULT: 'http://127.0.0.1:9222'
name	Server name TYPE: str DEFAULT: 'chrome-devtools'
**kwargs	Additional MCPServerConfig parameters DEFAULT: {}

RETURNS	DESCRIPTION
List[str]	List of registered tool names

Source code in packages/ai-parrot/src/parrot/mcp/integration.py

async def add_chrome_devtools_mcp_server(
    self,
    browser_url: str = "http://127.0.0.1:9222",
    name: str = "chrome-devtools",
    **kwargs
) -> List[str]:
    """Add Chrome DevTools MCP server capability.

    Args:
        browser_url: URL where Chrome is listening for devtools protocol
        name: Server name
        **kwargs: Additional MCPServerConfig parameters

    Returns:
        List of registered tool names
    """
    config = create_chrome_devtools_mcp_server(
        browser_url=browser_url,
        name=name,
        **kwargs
    )
    return await self.add_mcp_server(config)

add_google_maps_mcp_server async

add_google_maps_mcp_server(name: str = 'google-maps', **kwargs) -> List[str]

Add Google Maps MCP server capability.

PARAMETER	DESCRIPTION
name	Server name TYPE: str DEFAULT: 'google-maps'
**kwargs	Additional MCPServerConfig parameters DEFAULT: {}

RETURNS	DESCRIPTION
List[str]	List of registered tool names

Source code in packages/ai-parrot/src/parrot/mcp/integration.py

async def add_google_maps_mcp_server(
    self,
    name: str = "google-maps",
    **kwargs
) -> List[str]:
    """Add Google Maps MCP server capability.

    Args:
        name: Server name
        **kwargs: Additional MCPServerConfig parameters

    Returns:
        List of registered tool names
    """
    config = create_google_maps_mcp_server(
        name=name,
        **kwargs
    )
    return await self.add_mcp_server(config)

add_quic_mcp_server async

add_quic_mcp_server(name: str, host: str, port: int, cert_path: Optional[str] = None, **kwargs) -> List[str]

Add a QUIC/HTTP3 MCP server connection.

Source code in packages/ai-parrot/src/parrot/mcp/integration.py

async def add_quic_mcp_server(
    self,
    name: str,
    host: str,
    port: int,
    cert_path: Optional[str] = None,
    **kwargs
) -> List[str]:
    """Add a QUIC/HTTP3 MCP server connection."""
    config = create_quic_mcp_server(name, host, port, cert_path, **kwargs)
    return await self.add_mcp_server(config)

add_websocket_mcp_server async

add_websocket_mcp_server(name: str, url: str, auth_type: Optional[str] = None, auth_config: Optional[Dict[str, Any]] = None, headers: Optional[Dict[str, str]] = None, **kwargs) -> List[str]

Add a WebSocket MCP server connection.

PARAMETER	DESCRIPTION
name	Server name TYPE: str
url	WebSocket URL (ws:// or wss://) TYPE: str
auth_type	Authentication type ("bearer", "api_key", "oauth") TYPE: Optional[str] DEFAULT: None
auth_config	Authentication configuration TYPE: Optional[Dict[str, Any]] DEFAULT: None
headers	Additional headers for WebSocket upgrade TYPE: Optional[Dict[str, str]] DEFAULT: None
**kwargs	Additional MCPServerConfig parameters DEFAULT: {}

RETURNS	DESCRIPTION
List[str]	List of registered tool names

Example

await agent.add_websocket_mcp_server( ... "my-ws-server", ... "ws://localhost:8766/mcp/ws", ... auth_type="bearer", ... auth_config={"token": "my-token"} ... )

Source code in packages/ai-parrot/src/parrot/mcp/integration.py

async def add_websocket_mcp_server(
    self,
    name: str,
    url: str,
    auth_type: Optional[str] = None,
    auth_config: Optional[Dict[str, Any]] = None,
    headers: Optional[Dict[str, str]] = None,
    **kwargs
) -> List[str]:
    """Add a WebSocket MCP server connection.

    Args:
        name: Server name
        url: WebSocket URL (ws:// or wss://)
        auth_type: Authentication type ("bearer", "api_key", "oauth")
        auth_config: Authentication configuration
        headers: Additional headers for WebSocket upgrade
        **kwargs: Additional MCPServerConfig parameters

    Returns:
        List of registered tool names

    Example:
        >>> await agent.add_websocket_mcp_server(
        ...     "my-ws-server",
        ...     "ws://localhost:8766/mcp/ws",
        ...     auth_type="bearer",
        ...     auth_config={"token": "my-token"}
        ... )
    """
    config = create_websocket_mcp_server(
        name, url, auth_type, auth_config, headers, **kwargs
    )
    return await self.add_mcp_server(config)

reconfigure_mcp_server async

reconfigure_mcp_server(config: MCPClientConfig) -> List[str]

Reconfigure an existing MCP server with new configuration.

PARAMETER	DESCRIPTION
config	New MCPServerConfig with updated parameters TYPE: MCPClientConfig

RETURNS	DESCRIPTION
List[str]	List of registered tool names

Source code in packages/ai-parrot/src/parrot/mcp/integration.py

async def reconfigure_mcp_server(self, config: MCPServerConfig) -> List[str]:
    """Reconfigure an existing MCP server with new configuration.

    Args:
        config: New MCPServerConfig with updated parameters

    Returns:
        List of registered tool names
    """
    return await self.tool_manager.reconfigure_mcp_server(config)

reconfigure_fireflies_mcp_server async

reconfigure_fireflies_mcp_server(api_key: str, **kwargs) -> List[str]

Reconfigure Fireflies MCP server with a new API key.

This is useful in multi-user scenarios where each user provides their own Fireflies API key. The method will disconnect the existing connection and reconnect with the new credentials.

PARAMETER	DESCRIPTION
api_key	New Fireflies API key TYPE: str
**kwargs	Additional MCPServerConfig parameters DEFAULT: {}

RETURNS	DESCRIPTION
List[str]	List of registered tool names

Example

Initial setup with user 1's API key

await agent.add_fireflies_mcp_server(api_key="user1-api-key")

Later, reconfigure with user 2's API key

await agent.reconfigure_fireflies_mcp_server(api_key="user2-api-key")

Source code in packages/ai-parrot/src/parrot/mcp/integration.py

async def reconfigure_fireflies_mcp_server(self, api_key: str, **kwargs) -> List[str]:
    """Reconfigure Fireflies MCP server with a new API key.

    This is useful in multi-user scenarios where each user provides their own
    Fireflies API key. The method will disconnect the existing connection and
    reconnect with the new credentials.

    Args:
        api_key: New Fireflies API key
        **kwargs: Additional MCPServerConfig parameters

    Returns:
        List of registered tool names

    Example:
        >>> # Initial setup with user 1's API key
        >>> await agent.add_fireflies_mcp_server(api_key="user1-api-key")

        >>> # Later, reconfigure with user 2's API key
        >>> await agent.reconfigure_fireflies_mcp_server(api_key="user2-api-key")
    """
    config = create_fireflies_mcp_server(api_key=api_key, **kwargs)
    return await self.reconfigure_mcp_server(config)

reconfigure_perplexity_mcp_server async

reconfigure_perplexity_mcp_server(api_key: str, name: str = 'perplexity', **kwargs) -> List[str]

Reconfigure Perplexity MCP server with a new API key.

Useful for updating the API key without restarting the agent.

PARAMETER	DESCRIPTION
api_key	New Perplexity API key TYPE: str
name	Server name (default: "perplexity") TYPE: str DEFAULT: 'perplexity'
**kwargs	Additional MCPServerConfig parameters DEFAULT: {}

RETURNS	DESCRIPTION
List[str]	List of registered tool names

Source code in packages/ai-parrot/src/parrot/mcp/integration.py

async def reconfigure_perplexity_mcp_server(self, api_key: str, name: str = "perplexity", **kwargs) -> List[str]:
    """Reconfigure Perplexity MCP server with a new API key.

    Useful for updating the API key without restarting the agent.

    Args:
        api_key: New Perplexity API key
        name: Server name (default: "perplexity")
        **kwargs: Additional MCPServerConfig parameters

    Returns:
        List of registered tool names
    """
    config = create_perplexity_mcp_server(api_key, name=name, **kwargs)
    return await self.reconfigure_mcp_server(config)

get_openai_mcp_tools

get_openai_mcp_tools(server_names: Optional[List[str]] = None) -> List[Dict[str, Any]]

Get OpenAI-compatible MCP definitions for registered servers.

PARAMETER	DESCRIPTION
server_names	Optional list of server names to filter by TYPE: Optional[List[str]] DEFAULT: None

RETURNS	DESCRIPTION
List[Dict[str, Any]]	List of tool definitions in valid OpenAI MCP format

Source code in packages/ai-parrot/src/parrot/mcp/integration.py

def get_openai_mcp_tools(self, server_names: Optional[List[str]] = None) -> List[Dict[str, Any]]:
    """Get OpenAI-compatible MCP definitions for registered servers.

    Args:
        server_names: Optional list of server names to filter by

    Returns:
        List of tool definitions in valid OpenAI MCP format
    """
    return self.tool_manager.get_openai_mcp_definitions(server_names)

add_alphavantage_mcp_server async

add_alphavantage_mcp_server(api_key: Optional[str] = None, name: str = 'alphavantage', **kwargs) -> List[str]

Add AlphaVantage MCP server capability.

PARAMETER	DESCRIPTION
api_key	AlphaVantage API key TYPE: Optional[str] DEFAULT: None
name	Server name (default: "alphavantage") TYPE: str DEFAULT: 'alphavantage'
**kwargs	Additional MCPServerConfig parameters DEFAULT: {}

RETURNS	DESCRIPTION
List[str]	List of registered tool names

Source code in packages/ai-parrot/src/parrot/mcp/integration.py

async def add_alphavantage_mcp_server(
    self,
    api_key: Optional[str] = None,
    name: str = "alphavantage",
    **kwargs
) -> List[str]:
    """Add AlphaVantage MCP server capability.

    Args:
        api_key: AlphaVantage API key
        name: Server name (default: "alphavantage")
        **kwargs: Additional MCPServerConfig parameters

    Returns:
        List of registered tool names
    """
    config = create_alphavantage_mcp_server(api_key, name=name, **kwargs)
    return await self.add_mcp_server(config)

add_netsuite_mcp_server async

add_netsuite_mcp_server(account_id: str, client_id: str, user_id: str, **kwargs) -> List[str]

Add NetSuite MCP server capability via OAuth2 Authorization Code + PKCE.

Constructs the NetSuite MCP configuration from the given credentials and registers it with this agent. Scope is fixed to ["mcp"].

PARAMETER	DESCRIPTION
account_id	NetSuite account ID (e.g. "4984231"). TYPE: str
client_id	OAuth2 client ID from the NetSuite integration record. TYPE: str
user_id	Caller's user identifier for token storage scoping. TYPE: str
**kwargs	Additional keyword arguments forwarded to :func:create_netsuite_mcp_server (e.g. token_store, redirect_host, redirect_port). DEFAULT: {}

RETURNS	DESCRIPTION
List[str]	List of registered tool names from the NetSuite MCP server.

Example

tools = await agent.add_netsuite_mcp_server( ... account_id="4984231", ... client_id="my-client-id", ... user_id="user@co.com", ... )

Source code in packages/ai-parrot/src/parrot/mcp/integration.py

async def add_netsuite_mcp_server(
    self,
    account_id: str,
    client_id: str,
    user_id: str,
    **kwargs,
) -> List[str]:
    """Add NetSuite MCP server capability via OAuth2 Authorization Code + PKCE.

    Constructs the NetSuite MCP configuration from the given credentials and
    registers it with this agent. Scope is fixed to ``["mcp"]``.

    Args:
        account_id: NetSuite account ID (e.g. ``"4984231"``).
        client_id: OAuth2 client ID from the NetSuite integration record.
        user_id: Caller's user identifier for token storage scoping.
        **kwargs: Additional keyword arguments forwarded to
            :func:`create_netsuite_mcp_server` (e.g. ``token_store``,
            ``redirect_host``, ``redirect_port``).

    Returns:
        List of registered tool names from the NetSuite MCP server.

    Example:
        >>> tools = await agent.add_netsuite_mcp_server(
        ...     account_id="4984231",
        ...     client_id="my-client-id",
        ...     user_id="user@co.com",
        ... )
    """
    config = create_netsuite_mcp_server(
        account_id=account_id,
        client_id=client_id,
        user_id=user_id,
        **kwargs,
    )
    return await self.add_mcp_server(config)

add_netsuite_m2m_mcp_server async

add_netsuite_m2m_mcp_server(account_id: str, client_id: str, certificate_id: str, private_key_path: str, **kwargs) -> List[str]

Add NetSuite MCP server via OAuth2 Client Credentials (M2M) with certificate.

No browser login is needed — the agent authenticates using a JWT signed with the provided private key.

PARAMETER	DESCRIPTION
account_id	NetSuite account ID (e.g. "4984231"). TYPE: str
client_id	OAuth2 client ID from the NetSuite integration record. TYPE: str
certificate_id	Mapping Key from the uploaded certificate in NetSuite. TYPE: str
private_key_path	Path to a PEM-encoded RSA private key file. TYPE: str
**kwargs	Forwarded to :func:create_netsuite_m2m_mcp_server. DEFAULT: {}

RETURNS	DESCRIPTION
List[str]	List of registered tool names.

Example

tools = await agent.add_netsuite_m2m_mcp_server( ... account_id="4984231", ... client_id="abc...", ... certificate_id="XYZ123", ... private_key_path="/path/to/private.pem", ... )

Source code in packages/ai-parrot/src/parrot/mcp/integration.py

async def add_netsuite_m2m_mcp_server(
    self,
    account_id: str,
    client_id: str,
    certificate_id: str,
    private_key_path: str,
    **kwargs,
) -> List[str]:
    """Add NetSuite MCP server via OAuth2 Client Credentials (M2M) with certificate.

    No browser login is needed — the agent authenticates using a JWT
    signed with the provided private key.

    Args:
        account_id: NetSuite account ID (e.g. ``"4984231"``).
        client_id: OAuth2 client ID from the NetSuite integration record.
        certificate_id: Mapping Key from the uploaded certificate in NetSuite.
        private_key_path: Path to a PEM-encoded RSA private key file.
        **kwargs: Forwarded to :func:`create_netsuite_m2m_mcp_server`.

    Returns:
        List of registered tool names.

    Example:
        >>> tools = await agent.add_netsuite_m2m_mcp_server(
        ...     account_id="4984231",
        ...     client_id="abc...",
        ...     certificate_id="XYZ123",
        ...     private_key_path="/path/to/private.pem",
        ... )
    """
    cfg = create_netsuite_m2m_mcp_server(
        account_id=account_id,
        client_id=client_id,
        certificate_id=certificate_id,
        private_key_path=private_key_path,
        **kwargs,
    )
    if hasattr(cfg, "_ensure_oauth_token"):
        await cfg._ensure_oauth_token()
    return await self.add_mcp_server(cfg)

add_genmedia_mcp_servers async

add_genmedia_mcp_servers(**kwargs) -> Dict[str, List[str]]

Add all Google GenMedia MCP servers.

Available servers: - mcp-avtool-go - mcp-chirp3-go - mcp-gemini-go - mcp-imagen-go - mcp-lyria-go - mcp-veo-go

Source code in packages/ai-parrot/src/parrot/mcp/integration.py

async def add_genmedia_mcp_servers(
    self,
    **kwargs
) -> Dict[str, List[str]]:
    """
    Add all Google GenMedia MCP servers.

    Available servers:
    - mcp-avtool-go
    - mcp-chirp3-go
    - mcp-gemini-go
    - mcp-imagen-go
    - mcp-lyria-go
    - mcp-veo-go
    """
    project_id = config.get('PROJECT_ID')
    location = config.get('LOCATION', 'us-central1')

    if not project_id:
        self.logger.warning("PROJECT_ID not found in config. GenMedia servers might fail.")

    servers = [
        "mcp-avtool-go",
        "mcp-chirp3-go",
        "mcp-gemini-go",
        "mcp-imagen-go",
        "mcp-lyria-go",
        "mcp-veo-go"
    ]

    results = {}

    for server_bin in servers:
        try:
            # Remove 'mcp-' prefix and '-go' suffix for the name if desired,
            # or just use the binary name. Let's use a cleaner name.
            name = server_bin.replace("mcp-", "").replace("-go", "")

            server_config = MCPServerConfig(
                name=name,
                command=server_bin,  # Binary name in PATH
                args=[],
                transport="stdio",
                env={
                    "MCP_SERVER_REQUEST_TIMEOUT": "55000",
                    "PROJECT_ID": project_id,  # Will be filtered if None by StdioMCPSession
                    "LOCATION": location
                }
            )

            tools = await self.add_mcp_server(server_config)
            results[name] = tools
        except Exception as e:
            self.logger.error(
                f"Failed to add GenMedia server {server_bin}: {e}"
            )
            results[name] = []

    return results

setup_mcp_servers async

setup_mcp_servers(configurations: List[MCPClientConfig]) -> None

Setup multiple MCP servers during initialization.

This is useful for configuring an agent with multiple MCP servers at once, typically during agent creation or from configuration files.

PARAMETER	DESCRIPTION
configurations	List of MCPServerConfig objects TYPE: List[MCPClientConfig]

Source code in packages/ai-parrot/src/parrot/mcp/integration.py

async def setup_mcp_servers(self, configurations: List[MCPServerConfig]) -> None:
    """
    Setup multiple MCP servers during initialization.

    This is useful for configuring an agent with multiple MCP servers
    at once, typically during agent creation or from configuration files.

    Args:
        configurations: List of MCPServerConfig objects
    """
    for config in configurations:
        try:
            tools = await self.add_mcp_server(config)
            # Check for logger, fallback to print if not available
            if hasattr(self, 'logger'):
                self.logger.info(
                    f"Added MCP server '{config.name}' with tools: {tools}"
                )
        except Exception as e:
            if hasattr(self, 'logger'):
                self.logger.error(
                    f"Failed to add MCP server '{config.name}': {e}",
                    exc_info=True
                )

MCPServerConfig dataclass

MCPServerConfig(name: str, url: Optional[str] = None, command: Optional[str] = None, args: Optional[List[str]] = None, env: Optional[Dict[str, str]] = None, description: Optional[str] = None, auth_credential: Optional[AuthCredential] = None, auth_type: Optional[AuthScheme] = None, auth_config: Dict[str, Any] = dict(), token_supplier: Optional[Callable[[], Optional[str]]] = None, transport: str = 'auto', base_path: Optional[str] = None, events_path: Optional[str] = None, socket_path: Optional[str] = None, headers: Dict[str, str] = dict(), header_provider: Optional[Callable[[ReadonlyContext], Dict[str, str]]] = None, tool_filter: Optional[Union[List[str], Callable[[ToolPredicate], bool]]] = None, allowed_tools: Optional[List[str]] = None, blocked_tools: Optional[List[str]] = None, require_confirmation: Union[bool, Callable[[str, Dict[str, Any]], bool]] = False, tool_name_prefix: Optional[str] = None, timeout: float = 30.0, retry_count: int = 3, startup_delay: float = 0.5, rate_limit_max_retries: int = 2, rate_limit_max_wait: float = 60.0, rate_limit_base_delay: float = 1.0, kill_timeout: float = 5.0, quic_config: Any = None, oauth2: Optional[MCPOAuth2Config] = None, auth_preset: Optional[str] = None, user_id: Optional[str] = None, inject_broker_credential: bool = False)

Complete configuration for external MCP server connection.

Supports both static configuration and dynamic behavior through header_provider and token_supplier callbacks.

Example

Static config

config = MCPClientConfig( ... name="my-server", ... url="http://localhost:8080/mcp", ... transport="http", ... headers={"X-API-Key": "secret"} ... )

Dynamic headers based on context

def my_header_provider(ctx): ... return {"X-User-ID": ctx.user_id} if ctx else {} config = MCPClientConfig( ... name="my-server", ... url="http://localhost:8080/mcp", ... header_provider=my_header_provider ... )

get_headers async

get_headers(context: Optional[ReadonlyContext] = None) -> Dict[str, str]

Get merged static, auth, and dynamic headers.

Order of precedence (later overrides earlier): 1. Static headers from self.headers 2. Auth headers from auth_credential.get_auth_headers() 3. Dynamic headers from header_provider(context)

PARAMETER	DESCRIPTION
context	Optional ReadonlyContext for dynamic header generation TYPE: Optional[ReadonlyContext] DEFAULT: None

RETURNS	DESCRIPTION
Dict[str, str]	Merged dictionary of all headers

Example

headers = await config.get_headers(ctx)

Returns:

Source code in packages/ai-parrot/src/parrot/mcp/client.py

async def get_headers(self, context: Optional['ReadonlyContext'] = None) -> Dict[str, str]:
    """Get merged static, auth, and dynamic headers.

    Order of precedence (later overrides earlier):
    1. Static headers from self.headers
    2. Auth headers from auth_credential.get_auth_headers()
    3. Dynamic headers from header_provider(context)

    Args:
        context: Optional ReadonlyContext for dynamic header generation

    Returns:
        Merged dictionary of all headers

    Example:
        >>> headers = await config.get_headers(ctx)
        >>> # Returns: {"X-API-Key": "...", "Authorization": "Bearer ...", "X-User-ID": "123"}
    """
    result = dict(self.headers)

    # Add auth headers from auth_credential only when oauth2 is NOT configured.
    # When oauth2 is set the MCP SDK handles authentication at the transport layer.
    if self.auth_credential and not self.oauth2:
        auth_headers = self.auth_credential.get_auth_headers()
        result |= auth_headers

    # Add dynamic headers from provider
    if self.header_provider and context:
        dynamic = self.header_provider(context)
        # Support both sync and async header providers
        if asyncio.iscoroutine(dynamic):
            dynamic = await dynamic
        result.update(dynamic)

    # FEAT-264: Inject the CredentialBroker-resolved per-user bearer token
    # when inject_broker_credential=True.  The token lives in the per-call
    # ContextVar set by the tool-loop seam (AbstractTool.execute) so this
    # path is only active during a credentialed tool invocation.
    if self.inject_broker_credential:
        try:
            from parrot.tools.abstract import current_credential
            cred = current_credential()
            if cred is not None:
                # Never overwrite an existing Authorization header that was
                # set by a higher-priority source (auth_credential, header_provider).
                result.setdefault("Authorization", f"Bearer {cred}")
        except ImportError:
            pass  # parrot.tools.abstract not importable in edge environments

    return result

validate_transport

validate_transport() -> None

Validate transport-specific configuration.

RAISES	DESCRIPTION
ValueError	If configuration is invalid for the specified transport

Source code in packages/ai-parrot/src/parrot/mcp/client.py

def validate_transport(self) -> None:
    """Validate transport-specific configuration.

    Raises:
        ValueError: If configuration is invalid for the specified transport
    """
    if self.transport == "stdio" and not self.command:
        raise ValueError("stdio transport requires 'command' field")
    if self.transport == "http" and not self.url:
        raise ValueError("http transport requires 'url' field")
    if self.transport == "sse" and not self.url:
        raise ValueError("sse transport requires 'url' field")
    if self.transport == "unix" and not self.socket_path:
        raise ValueError("unix transport requires 'socket_path' field")
    if self.transport == "websocket" and not self.url:
        raise ValueError("websocket transport requires 'url' field")

from_yaml_config classmethod

from_yaml_config(config_dict: Dict[str, Any], config_abs_path: str = '') -> MCPClientConfig

Load from YAML configuration with validation.

PARAMETER	DESCRIPTION
config_dict	Dictionary loaded from YAML file TYPE: Dict[str, Any]
config_abs_path	Absolute path to YAML file (for error messages) TYPE: str DEFAULT: ''

RETURNS	DESCRIPTION
MCPClientConfig	MCPClientConfig instance

RAISES	DESCRIPTION
ValueError	If configuration is invalid

Example

import yaml with open("mcp_servers.yaml") as f: ... config = yaml.safe_load(f) mcp_config = MCPClientConfig.from_yaml_config( ... config['servers']['my-server'], ... "/path/to/mcp_servers.yaml" ... )

Source code in packages/ai-parrot/src/parrot/mcp/client.py

@classmethod
def from_yaml_config(
    cls,
    config_dict: Dict[str, Any],
    config_abs_path: str = ""
) -> 'MCPClientConfig':
    """Load from YAML configuration with validation.

    Args:
        config_dict: Dictionary loaded from YAML file
        config_abs_path: Absolute path to YAML file (for error messages)

    Returns:
        MCPClientConfig instance

    Raises:
        ValueError: If configuration is invalid

    Example:
        >>> import yaml
        >>> with open("mcp_servers.yaml") as f:
        ...     config = yaml.safe_load(f)
        >>> mcp_config = MCPClientConfig.from_yaml_config(
        ...     config['servers']['my-server'],
        ...     "/path/to/mcp_servers.yaml"
        ... )
    """
    # Validate transport selection - exactly one transport indicator must be set
    transport_fields = {
        'command': config_dict.get('command'),
        'url': config_dict.get('url'),
        'socket_path': config_dict.get('socket_path'),
    }
    populated = [k for k, v in transport_fields.items() if v is not None]

    if not populated:
        raise ValueError(
            f"At least one of [command, url, socket_path] must be set in {config_abs_path}"
        )
    if len(populated) > 1 and config_dict.get('transport', 'auto') == 'auto':
        raise ValueError(
            f"Exactly one of [command, url, socket_path] should be set for auto transport. "
            f"Got: {populated}. Set 'transport' explicitly to override."
        )

    # Handle auth_credential if present as dict
    if 'auth_credential' in config_dict and isinstance(config_dict['auth_credential'], dict):
        config_dict['auth_credential'] = AuthCredential(**config_dict['auth_credential'])

    # Handle OAuth2 preset and inline oauth2 config (FEAT-262)
    auth_preset = config_dict.pop('auth_preset', None)
    oauth2_dict = config_dict.pop('oauth2', None)

    if auth_preset:
        preset = get_mcp_oauth2_preset(auth_preset)
        if not preset:
            raise ValueError(
                f"Unknown MCP OAuth2 preset: '{auth_preset}' in {config_abs_path}"
            )
        base = preset.model_dump(
            exclude_none=True,
            exclude={'name', 'display_name', 'url_template', 'required_params'},
        )
        if oauth2_dict:
            base.update(oauth2_dict)  # inline oauth2 dict overrides preset defaults
        config_dict['oauth2'] = MCPOAuth2Config(**base)
    elif oauth2_dict:
        config_dict['oauth2'] = MCPOAuth2Config(**oauth2_dict)

    config_dict['auth_preset'] = auth_preset

    # Filter to only known fields
    known_fields = {f.name for f in cls.__dataclass_fields__.values()}  # pylint: disable=no-member
    filtered = {k: v for k, v in config_dict.items() if k in known_fields}

    instance = cls(**filtered)
    instance.validate_transport()
    return instance

MCPClient

MCPClient(config: MCPClientConfig, tool_name_prefix: Optional[str] = None)

Complete MCP client with stdio and HTTP transport support.

Source code in packages/ai-parrot/src/parrot/mcp/integration.py

def __init__(
    self,
    config: MCPServerConfig,
    tool_name_prefix: Optional[str] = None
):
    self.tool_name_prefix = tool_name_prefix or config.tool_name_prefix or f"mcp_{config.name}"
    self.config = config
    self.logger = logging.getLogger(f"MCPClient.{config.name}")
    self._session = None
    self._connected = False
    self._available_tools = []

connect async

connect()

Connect to MCP server using appropriate transport.

Source code in packages/ai-parrot/src/parrot/mcp/integration.py

async def connect(self):
    """Connect to MCP server using appropriate transport."""
    if self._connected:
        return

    transport = self._detect_transport()

    try:
        if transport == "stdio":
            self._session = StdioMCPSession(self.config, self.logger)
        elif transport == "http":
            self._session = HttpMCPSession(self.config, self.logger)
        elif transport == "sse":
            self._session = SseMCPSession(self.config, self.logger)
        elif transport == "unix":
            self._session = UnixMCPSession(self.config, self.logger)
        elif transport == "websocket":

            self._session = WebSocketMCPSession(self.config, self.logger)
        elif transport == "quic":
            try:
                self._session = QuicMCPSession(self.config, self.logger)
            except ImportError as e:
                raise ImportError(
                    "QUIC transport requires 'aioquic' package. Install with: pip install aioquic msgpack"
                ) from e
        else:
            raise ValueError(
                f"Unsupported transport: {transport}"
            )

        await self._session.connect()
        self._available_tools = await self._session.list_tools()
        self._connected = True

        self.logger.info(
            f"Connected to MCP server {self.config.name} "
            f"via {transport} with {len(self._available_tools)} tools"
        )

    except Exception as e:
        self.logger.error(f"Failed to connect: {e}")
        await self.disconnect()
        raise

call_tool async

call_tool(tool_name: str, arguments: Dict[str, Any], headers: Optional[Dict[str, str]] = None)

Call an MCP tool.

Rate-limit responses (-32429) are retried with the server-suggested retryAfter delay (or an exponential fallback), up to rate_limit_max_retries. A suggested wait longer than rate_limit_max_wait fails fast instead of blocking the agent loop — the error is surfaced so the caller can move on rather than hammering a throttled server.

Source code in packages/ai-parrot/src/parrot/mcp/integration.py

async def call_tool(
    self,
    tool_name: str,
    arguments: Dict[str, Any],
    headers: Optional[Dict[str, str]] = None
):
    """Call an MCP tool.

    Rate-limit responses (``-32429``) are retried with the server-suggested
    ``retryAfter`` delay (or an exponential fallback), up to
    ``rate_limit_max_retries``. A suggested wait longer than
    ``rate_limit_max_wait`` fails fast instead of blocking the agent loop —
    the error is surfaced so the caller can move on rather than hammering a
    throttled server.
    """
    if not self._connected:
        raise MCPConnectionError("Not connected to MCP server")

    max_retries = getattr(self.config, "rate_limit_max_retries", 2)
    max_wait = getattr(self.config, "rate_limit_max_wait", 60.0)
    base_delay = getattr(self.config, "rate_limit_base_delay", 1.0)

    attempt = 0
    while True:
        try:
            return await self._session.call_tool(tool_name, arguments)
        except MCPRateLimitError as exc:
            delay = (
                exc.retry_after
                if exc.retry_after is not None
                else base_delay * (2 ** attempt)
            )
            if attempt >= max_retries or delay > max_wait:
                self.logger.warning(
                    "Rate limit on tool '%s'; not retrying "
                    "(attempt %d/%d, suggested wait %.1fs, cap %.1fs): %s",
                    tool_name, attempt + 1, max_retries + 1,
                    delay, max_wait, exc,
                )
                raise
            self.logger.info(
                "Rate limited on tool '%s'; backing off %.1fs before retry %d/%d",
                tool_name, delay, attempt + 1, max_retries,
            )
            await asyncio.sleep(delay)
            attempt += 1

get_available_tools async

get_available_tools() -> List[Dict[str, Any]]

Get raw available tools from server.

Source code in packages/ai-parrot/src/parrot/mcp/integration.py

async def get_available_tools(self) -> List[Dict[str, Any]]:
    """Get raw available tools from server."""
    if not self._connected:
        raise MCPConnectionError("Not connected to MCP server")

    tools = []
    for tool in self._available_tools:
        tool_dict = {
            'name': getattr(tool, 'name', 'unknown'),
            'description': getattr(tool, 'description', ''),
            'inputSchema': getattr(tool, 'inputSchema', {})
        }
        tools.append(tool_dict)
    return tools

get_tools_for_context

get_tools_for_context(context: Optional[ReadonlyContext] = None) -> List[Dict[str, Any]]

Get tools, filtered by context.

If a context is provided, tools will be filtered based on the context's scopes and roles. Tools may specify required scopes in their metadata.

PARAMETER	DESCRIPTION
context	Optional ReadonlyContext for filtering TYPE: Optional[ReadonlyContext] DEFAULT: None

RETURNS	DESCRIPTION
List[Dict[str, Any]]	List of tool dictionaries accessible to the context

Example

from parrot.mcp.context import ReadonlyContext ctx = ReadonlyContext( ... agent_id="my-agent", ... scopes=["read:data", "write:data"] ... ) tools = client.get_tools_for_context(ctx)

Source code in packages/ai-parrot/src/parrot/mcp/integration.py

def get_tools_for_context(
    self,
    context: Optional['ReadonlyContext'] = None
) -> List[Dict[str, Any]]:
    """Get tools, filtered by context.

    If a context is provided, tools will be filtered based on the
    context's scopes and roles. Tools may specify required scopes
    in their metadata.

    Args:
        context: Optional ReadonlyContext for filtering

    Returns:
        List of tool dictionaries accessible to the context

    Example:
        >>> from parrot.mcp.context import ReadonlyContext
        >>> ctx = ReadonlyContext(
        ...     agent_id="my-agent",
        ...     scopes=["read:data", "write:data"]
        ... )
        >>> tools = client.get_tools_for_context(ctx)
    """
    all_tools = self.get_available_tools()

    if not context:
        return all_tools

    return [t for t in all_tools if self._can_access(t, context)]

get_tools async

get_tools(context: Optional[ReadonlyContext] = None) -> List[MCPToolProxy]

Get tools filtered by configuration and context.

Filtering precedence: 1. tool_filter (new dynamic filtering) - highest priority 2. allowed_tools / blocked_tools (legacy) - fallback 3. No filter - all tools

PARAMETER	DESCRIPTION
context	Optional ReadonlyContext for context-aware decisions TYPE: Optional[ReadonlyContext] DEFAULT: None

RETURNS	DESCRIPTION
List[MCPToolProxy]	List of MCPToolProxy objects that are available

Example

Get all tools

tools = await client.get_tools()

Get tools filtered by context

ctx = ReadonlyContext(user_id="user123", roles=["admin"]) admin_tools = await client.get_tools(context=ctx)

Source code in packages/ai-parrot/src/parrot/mcp/integration.py

async def get_tools(
    self,
    context: Optional['ReadonlyContext'] = None
) -> List[MCPToolProxy]:
    """Get tools filtered by configuration and context.

    Filtering precedence:
    1. tool_filter (new dynamic filtering) - highest priority
    2. allowed_tools / blocked_tools (legacy) - fallback
    3. No filter - all tools

    Args:
        context: Optional ReadonlyContext for context-aware decisions

    Returns:
        List of MCPToolProxy objects that are available

    Example:
        >>> # Get all tools
        >>> tools = await client.get_tools()
        >>>
        >>> # Get tools filtered by context
        >>> ctx = ReadonlyContext(user_id="user123", roles=["admin"])
        >>> admin_tools = await client.get_tools(context=ctx)
    """
    available = await self.get_available_tools()

    # Apply tool_filter if configured
    if self.config.tool_filter:
        available = self._filter_tools(available, context)
    # Fallback to legacy allowed_tools/blocked_tools
    elif self.config.allowed_tools or self.config.blocked_tools:
        available = self._filter_tools_legacy(available)

    return available

disconnect async

disconnect()

Disconnect from MCP server.

Source code in packages/ai-parrot/src/parrot/mcp/integration.py

async def disconnect(self):
    """Disconnect from MCP server."""
    if not self._connected:
        return

    self._connected = False

    if self._session:
        await self._session.disconnect()
        self._session = None

    self._available_tools = []
    self.logger.info(f"Disconnected from {self.config.name}")

NetSuiteM2MAuth

NetSuiteM2MAuth(*, client_id: str, certificate_id: str, private_key_path: str, account_id: str, token_url: str | None = None, scopes: list[str] | None = None, token_store: TokenStore | None = None)

OAuth2 Client Credentials (M2M) for NetSuite using certificate-based JWT assertion.

NetSuite M2M requires a signed JWT as the client_assertion when requesting an access token. The JWT is signed with the private key whose matching X.509 certificate was uploaded to the NetSuite Integration Record.

PARAMETER	DESCRIPTION
client_id	OAuth2 client ID from the NetSuite integration record. TYPE: str
certificate_id	Certificate ID shown in NetSuite after uploading the public certificate. TYPE: str
private_key_path	Path to the PEM-encoded RSA private key file. TYPE: str
account_id	NetSuite account ID (e.g. "4984231"). TYPE: str
token_url	NetSuite token endpoint. Built automatically when None. TYPE: str | None DEFAULT: None
scopes	OAuth2 scopes (default ["mcp"]). TYPE: list[str] | None DEFAULT: None
token_store	Optional :class:TokenStore for persisting tokens. TYPE: TokenStore | None DEFAULT: None

Source code in packages/ai-parrot/src/parrot/mcp/oauth.py

def __init__(
    self,
    *,
    client_id: str,
    certificate_id: str,
    private_key_path: str,
    account_id: str,
    token_url: str | None = None,
    scopes: list[str] | None = None,
    token_store: TokenStore | None = None,
):
    self.client_id = client_id
    self.certificate_id = certificate_id
    self.account_id = account_id
    self.scopes = scopes or ["mcp"]
    self.token_store = token_store or InMemoryTokenStore()
    self._logger = logging.getLogger("NetSuiteM2MAuth")

    self.token_url = token_url or (
        f"https://{account_id}.suitetalk.api.netsuite.com"
        "/services/rest/auth/oauth2/v1/token"
    )

    from cryptography.hazmat.primitives import serialization
    with open(private_key_path, "rb") as f:
        self._private_key = serialization.load_pem_private_key(f.read(), password=None)

    self._token: dict | None = None
    self._server_name = "netsuite"

ensure_token async

ensure_token(user_id: str = 'm2m') -> str

Obtain or refresh an access token via Client Credentials grant.

Source code in packages/ai-parrot/src/parrot/mcp/oauth.py

async def ensure_token(self, user_id: str = "m2m") -> str:
    """Obtain or refresh an access token via Client Credentials grant."""
    stored = await self.token_store.get(user_id, self._server_name)
    if stored and stored.get("expires_at", 0) - _now() > 60:
        self._token = stored
        return stored["access_token"]

    assertion = self._build_assertion()
    async with ClientSession() as sess:
        data = {
            "grant_type": "client_credentials",
            "client_assertion_type": "urn:ietf:params:oauth:client-assertion-type:jwt-bearer",
            "client_assertion": assertion,
        }
        async with sess.post(self.token_url, data=data, timeout=30) as resp:
            tok = await resp.json()
            if resp.status != 200 or "access_token" not in tok:
                raise RuntimeError(
                    f"NetSuite M2M token exchange failed ({resp.status}): {tok}"
                )

    expires_in = int(tok.get("expires_in", 3600))
    self._token = {
        "access_token": tok["access_token"],
        "token_type": tok.get("token_type", "Bearer"),
        "expires_in": expires_in,
        "expires_at": _now() + expires_in,
        "scope": tok.get("scope"),
        "raw": tok,
    }
    await self.token_store.set(user_id, self._server_name, self._token)
    self._logger.info("NetSuite M2M token acquired (expires in %ds)", expires_in)
    return self._token["access_token"]

token_supplier

token_supplier() -> str | None

Synchronous hook called by the HTTP transport before each request.

Source code in packages/ai-parrot/src/parrot/mcp/oauth.py

def token_supplier(self) -> str | None:
    """Synchronous hook called by the HTTP transport before each request."""
    if not self._token:
        return None
    if self._token.get("expires_at", 0) - _now() < 60:
        return None
    return self._token.get("access_token")

TokenStore

Abstract token store interface.

InMemoryTokenStore

InMemoryTokenStore()

Bases: TokenStore

Simple in-memory token store (not persistent).

Source code in packages/ai-parrot/src/parrot/mcp/oauth.py

def __init__(self):
    self._data = {}

RedisTokenStore

RedisTokenStore(redis)

Bases: TokenStore

Redis-based token store.

Source code in packages/ai-parrot/src/parrot/mcp/oauth.py

def __init__(self, redis):
    self.redis = redis

VaultTokenStore

Bases: TokenStore

Vault-backed token store that encrypts OAuth tokens using AES-GCM.

Persists tokens in the DocumentDB Vault (via vault_utils) so they survive agent restarts. Falls back gracefully when the credential is not found or vault keys are unavailable.

The credential name follows the pattern mcp_oauth_{server_name}_{user_id}.

Example

store = VaultTokenStore() await store.set("user@co.com", "netsuite", token_dict) token = await store.get("user@co.com", "netsuite")

get async

get(user_id: str, server_name: str) -> Optional[Dict[str, Any]]

Retrieve a stored OAuth token from the Vault.

PARAMETER	DESCRIPTION
user_id	Owner's user identifier. TYPE: str
server_name	MCP server slug. TYPE: str

RETURNS	DESCRIPTION
Optional[Dict[str, Any]]	Decrypted token dict, or None if not found or vault unavailable.

Source code in packages/ai-parrot/src/parrot/mcp/oauth.py

async def get(self, user_id: str, server_name: str) -> Optional[Dict[str, Any]]:
    """Retrieve a stored OAuth token from the Vault.

    Args:
        user_id: Owner's user identifier.
        server_name: MCP server slug.

    Returns:
        Decrypted token dict, or ``None`` if not found or vault unavailable.
    """
    vault_name = self._vault_name(user_id, server_name)
    try:
        return await retrieve_vault_credential(user_id, vault_name)
    except KeyError:
        return None
    except RuntimeError as exc:
        self._logger.warning(
            "VaultTokenStore.get: vault keys unavailable for %s/%s — %s",
            user_id,
            server_name,
            exc,
        )
        return None

set async

set(user_id: str, server_name: str, token: Dict[str, Any]) -> None

Encrypt and persist an OAuth token in the Vault.

Degrades gracefully when vault keys are unavailable: logs a warning and returns without raising so the in-memory token remains usable.

PARAMETER	DESCRIPTION
user_id	Owner's user identifier. TYPE: str
server_name	MCP server slug. TYPE: str
token	Token dict to store (e.g. access_token, refresh_token, expires_at). TYPE: Dict[str, Any]

Source code in packages/ai-parrot/src/parrot/mcp/oauth.py

async def set(self, user_id: str, server_name: str, token: Dict[str, Any]) -> None:
    """Encrypt and persist an OAuth token in the Vault.

    Degrades gracefully when vault keys are unavailable: logs a warning
    and returns without raising so the in-memory token remains usable.

    Args:
        user_id: Owner's user identifier.
        server_name: MCP server slug.
        token: Token dict to store (e.g. access_token, refresh_token, expires_at).
    """
    vault_name = self._vault_name(user_id, server_name)
    try:
        await store_vault_credential(user_id, vault_name, token)
    except RuntimeError as exc:
        self._logger.warning(
            "VaultTokenStore.set: vault keys unavailable for %s/%s — %s",
            user_id,
            server_name,
            exc,
        )

delete async

delete(user_id: str, server_name: str) -> None

Remove a stored OAuth token from the Vault.

Tolerates a missing credential (already deleted) and vault unavailability — both are logged at warning level without raising.

PARAMETER	DESCRIPTION
user_id	Owner's user identifier. TYPE: str
server_name	MCP server slug. TYPE: str

Source code in packages/ai-parrot/src/parrot/mcp/oauth.py

async def delete(self, user_id: str, server_name: str) -> None:
    """Remove a stored OAuth token from the Vault.

    Tolerates a missing credential (already deleted) and vault
    unavailability — both are logged at warning level without raising.

    Args:
        user_id: Owner's user identifier.
        server_name: MCP server slug.
    """
    vault_name = self._vault_name(user_id, server_name)
    try:
        await delete_vault_credential(user_id, vault_name)
    except (KeyError, RuntimeError) as exc:
        self._logger.warning(
            "VaultTokenStore.delete: could not delete token for %s/%s — %s",
            user_id,
            server_name,
            exc,
        )

AuthScheme

Bases: str, Enum

Type-safe authentication schemes.

AuthCredential

Bases: BaseModel

Type-safe credential container with validation.

Validates that required fields are present based on the chosen scheme.

Example

Bearer token

cred = AuthCredential(scheme=AuthScheme.BEARER, token="my-token")

API Key with custom header

cred = AuthCredential( ... scheme=AuthScheme.API_KEY, ... api_key="secret", ... api_key_header="X-Custom-Key" ... )

Get headers

headers = cred.get_auth_headers()

validate_scheme_requirements

validate_scheme_requirements()

Validate that required fields are set for chosen scheme.

Source code in packages/ai-parrot/src/parrot/mcp/client.py

@model_validator(mode="after")
def validate_scheme_requirements(self):
    """Validate that required fields are set for chosen scheme."""
    scheme = self.scheme

    if scheme == AuthScheme.BEARER and not self.token:
        raise ValueError("Bearer scheme requires 'token' field")
    if scheme == AuthScheme.API_KEY and not self.api_key:
        raise ValueError("API Key scheme requires 'api_key' field")
    if scheme == AuthScheme.BASIC and (not self.username or not self.password):
        raise ValueError("Basic auth requires 'username' and 'password'")
    if scheme == AuthScheme.MTLS and (not self.cert_path or not self.key_path):
        raise ValueError("mTLS requires 'cert_path' and 'key_path'")
    if scheme == AuthScheme.AWS_SIG_V4 and (not self.aws_access_key or not self.aws_secret_key):
        raise ValueError("AWS Sig V4 requires 'aws_access_key' and 'aws_secret_key'")
    if scheme == AuthScheme.OAUTH2 and not self.token:
        raise ValueError("OAuth2 scheme requires 'token' field")

    return self

get_auth_headers

get_auth_headers() -> Dict[str, str]

Generate appropriate auth headers based on scheme.

RETURNS	DESCRIPTION
Dict[str, str]	Dictionary of authentication headers

Note

AWS Sig V4 and mTLS require special handling at the transport level and are not returned as simple headers.

Source code in packages/ai-parrot/src/parrot/mcp/client.py

def get_auth_headers(self) -> Dict[str, str]:
    """Generate appropriate auth headers based on scheme.

    Returns:
        Dictionary of authentication headers

    Note:
        AWS Sig V4 and mTLS require special handling at the transport level
        and are not returned as simple headers.
    """
    if self.scheme == AuthScheme.NONE or self.scheme is None:
        return {}
    elif self.scheme == AuthScheme.BEARER:
        return {"Authorization": f"Bearer {self.token}"}
    elif self.scheme == AuthScheme.API_KEY:
        value = f"Bearer {self.api_key}" if self.use_bearer_prefix else self.api_key
        return {self.api_key_header: value}
    elif self.scheme == AuthScheme.BASIC:
        creds = base64.b64encode(
            f"{self.username}:{self.password}".encode()
        ).decode()
        return {"Authorization": f"Basic {creds}"}
    elif self.scheme == AuthScheme.OAUTH2:
        return {"Authorization": f"Bearer {self.token}"}
    # AWS Sig V4 and mTLS are handled at transport level
    elif self.scheme in (AuthScheme.MTLS, AuthScheme.AWS_SIG_V4):
        return {}
    return {}

ReadonlyContext

Bases: BaseModel

Immutable context passed to tool operations.

This context provides information about the agent, user, and organizational context for tool execution. It enables: - Tool filtering based on user roles/scopes - Dynamic header generation - Multi-tenant isolation - Rate limiting by user/organization

Example

ctx = ReadonlyContext( ... agent_id="hr-agent", ... user_id="user-123", ... organization_id="acme-corp", ... roles=["admin", "hr"], ... scopes=["read:employees", "write:employees"] ... )

Context is immutable

ctx.user_id = "other" # Raises ValidationError

MCPSessionManager

MCPSessionManager(config: Any, max_retries: int = 3, base_wait: float = 2.0)

Manages session lifecycle and retry logic for MCP connections.

Features: - Session caching keyed by headers (for multi-user scenarios) - Automatic retry with exponential backoff on transient failures - Configurable max retries

Example

manager = MCPSessionManager(config, max_retries=3) session = await manager.create_session(headers={"X-User-ID": "123"})

Session is cached - same headers return same session

same_session = await manager.create_session(headers={"X-User-ID": "123"})

Source code in packages/ai-parrot/src/parrot/mcp/context.py

def __init__(
    self,
    config: Any,  # MCPClientConfig - avoid circular import
    max_retries: int = 3,
    base_wait: float = 2.0
):
    self.config = config
    self.max_retries = max_retries
    self.base_wait = base_wait
    self.sessions: Dict[int, Any] = {}
    self.logger = logging.getLogger(f"MCPSessionManager.{getattr(config, 'name', 'unknown')}")

create_session async

create_session(headers: Optional[Dict[str, str]] = None, force_new: bool = False) -> Any

Create or retrieve cached session with retry.

PARAMETER	DESCRIPTION
headers	Optional headers that affect session identity TYPE: Optional[Dict[str, str]] DEFAULT: None
force_new	If True, create new session even if cached TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
Any	MCP session object

Source code in packages/ai-parrot/src/parrot/mcp/context.py

async def create_session(
    self,
    headers: Optional[Dict[str, str]] = None,
    force_new: bool = False
) -> Any:
    """Create or retrieve cached session with retry.

    Args:
        headers: Optional headers that affect session identity
        force_new: If True, create new session even if cached

    Returns:
        MCP session object
    """
    session_key = self._session_key(headers)

    if not force_new and session_key in self.sessions:
        session = self.sessions[session_key]
        # Verify session is still valid
        if await self._is_session_valid(session):
            return session
        else:
            self.logger.debug("Cached session invalid, creating new one")
            del self.sessions[session_key]

    session = await self._create_session_with_retry(headers)
    self.sessions[session_key] = session
    return session

invalidate_session async

invalidate_session(headers: Optional[Dict[str, str]] = None) -> None

Invalidate a cached session.

Source code in packages/ai-parrot/src/parrot/mcp/context.py

async def invalidate_session(self, headers: Optional[Dict[str, str]] = None) -> None:
    """Invalidate a cached session."""
    session_key = self._session_key(headers)
    if session_key in self.sessions:
        session = self.sessions.pop(session_key)
        await self._close_session(session)

invalidate_all async

invalidate_all() -> None

Invalidate all cached sessions.

Source code in packages/ai-parrot/src/parrot/mcp/context.py

async def invalidate_all(self) -> None:
    """Invalidate all cached sessions."""
    for session in self.sessions.values():
        await self._close_session(session)
    self.sessions.clear()

TransientMCPError

Bases: Exception

Transient MCP errors that should be retried.

Use this for errors like connection timeouts, temporary server unavailability, rate limiting, etc. The retry_on_errors decorator will automatically retry operations that raise this exception.

MCPServerRegistry

Catalog of pre-built MCP server helpers available for user activation.

Wraps the module-level _REGISTRY list with lookup and validation methods. The registry is declarative — all entries are hand-authored above, not generated by reflection.

Example::

registry = MCPServerRegistry()
desc = registry.get_server("perplexity")
params = registry.validate_params("perplexity", {"api_key": "sk-..."})

list_servers

list_servers() -> List[MCPServerDescriptor]

Return all registered MCP server descriptors.

RETURNS	DESCRIPTION
List[MCPServerDescriptor]	List of :class:MCPServerDescriptor instances, one per helper.

Source code in packages/ai-parrot/src/parrot/mcp/registry.py

def list_servers(self) -> List[MCPServerDescriptor]:
    """Return all registered MCP server descriptors.

    Returns:
        List of :class:`MCPServerDescriptor` instances, one per helper.
    """
    return list(_REGISTRY)

get_server

get_server(name: str) -> Optional[MCPServerDescriptor]

Look up a single server descriptor by its registry slug.

PARAMETER	DESCRIPTION
name	Registry slug (e.g. "perplexity"). TYPE: str

RETURNS	DESCRIPTION
Optional[MCPServerDescriptor]	The matching :class:MCPServerDescriptor, or None if not found.

Source code in packages/ai-parrot/src/parrot/mcp/registry.py

def get_server(self, name: str) -> Optional[MCPServerDescriptor]:
    """Look up a single server descriptor by its registry slug.

    Args:
        name: Registry slug (e.g. ``"perplexity"``).

    Returns:
        The matching :class:`MCPServerDescriptor`, or ``None`` if not found.
    """
    for desc in _REGISTRY:
        if desc.name == name:
            return desc
    return None

validate_params

validate_params(name: str, params: Dict[str, Any]) -> Dict[str, Any]

Validate user-supplied parameters against the descriptor schema.

Checks that all required parameters are present. Fills in default values for optional parameters that are absent from params.

PARAMETER	DESCRIPTION
name	Registry slug of the server to validate against. TYPE: str
params	User-supplied parameter mapping. TYPE: Dict[str, Any]

RETURNS	DESCRIPTION
Dict[str, Any]	Cleaned parameter dict with defaults applied.

RAISES	DESCRIPTION
ValueError	If the server slug is not found, or if any required parameter is missing from params.

Source code in packages/ai-parrot/src/parrot/mcp/registry.py

def validate_params(
    self,
    name: str,
    params: Dict[str, Any],
) -> Dict[str, Any]:
    """Validate user-supplied parameters against the descriptor schema.

    Checks that all required parameters are present.  Fills in default
    values for optional parameters that are absent from ``params``.

    Args:
        name: Registry slug of the server to validate against.
        params: User-supplied parameter mapping.

    Returns:
        Cleaned parameter dict with defaults applied.

    Raises:
        ValueError: If the server slug is not found, or if any required
            parameter is missing from ``params``.
    """
    desc = self.get_server(name)
    if desc is None:
        raise ValueError(
            f"MCP server '{name}' not found in registry. "
            f"Available servers: {[s.name for s in _REGISTRY]}"
        )

    cleaned: Dict[str, Any] = dict(params)
    missing: List[str] = []

    for param in desc.params:
        if param.name not in cleaned:
            if param.required:
                missing.append(param.name)
            else:
                cleaned[param.name] = param.default

    if missing:
        raise ValueError(
            f"Missing required parameter(s) for MCP server '{name}': "
            + ", ".join(f"'{p}'" for p in missing)
        )

    return cleaned

MCPServerDescriptor

Bases: BaseModel

Catalog entry describing a single pre-built MCP server helper.

ATTRIBUTE	DESCRIPTION
name	Registry slug used as the identifier in API requests (e.g. "perplexity"). TYPE: str
display_name	Human-friendly label for UI display. TYPE: str
description	What the MCP server does. TYPE: str
method_name	Name of the MCPEnabledMixin method to call (e.g. "add_perplexity_mcp_server"). TYPE: str
params	Ordered list of accepted parameters. TYPE: List[MCPServerParam]
category	Grouping label for the catalog (e.g. "search", "media", "dev-tools"). TYPE: str
activatable	Whether the server can be activated via the POST endpoint. Servers without a create_* factory (e.g. genmedia) should set this to False. TYPE: bool

MCPServerParam

Bases: BaseModel

Describes a single parameter accepted by an MCP server helper.

ATTRIBUTE	DESCRIPTION
name	Parameter name (matches the Python keyword argument). TYPE: str
type	Expected value type; use SECRET for credentials. TYPE: MCPParamType
required	Whether the caller must supply a value. TYPE: bool
default	Default value used when required is False. TYPE: Optional[Any]
description	Human-readable explanation shown in the catalog. TYPE: str

MCPParamType

Bases: str, Enum

Type hint for an MCP server parameter.

The SECRET variant signals that the frontend should mask the input field and that the value must be stored in the Vault rather than persisted in DocumentDB plaintext.

UserMCPServerConfig

Bases: BaseModel

Persisted configuration for a user-activated MCP server.

This document is stored in the user_mcp_configs DocumentDB collection. Secret parameters (API keys, tokens) are never stored here — they live in the Vault; vault_credential_name points to the Vault entry.

ATTRIBUTE	DESCRIPTION
server_name	Registry slug of the activated server. TYPE: str
agent_id	Agent the server is scoped to. TYPE: str
user_id	Owner of this configuration. TYPE: str
params	Non-secret configuration parameters. TYPE: Dict[str, Any]
vault_credential_name	Name of the Vault credential that holds any secret values (None if no secrets are required). TYPE: Optional[str]
active	Soft-delete flag; False means this config is deactivated. TYPE: bool
created_at	ISO-8601 timestamp of initial creation. TYPE: Optional[str]
updated_at	ISO-8601 timestamp of last update. TYPE: Optional[str]

ActivateMCPServerRequest

Bases: BaseModel

Request body for the POST (activate) endpoint.

ATTRIBUTE	DESCRIPTION
server	Registry slug of the server to activate (e.g. "perplexity"). TYPE: str
params	All parameters, including secrets. The handler separates secret params before storing them in the Vault. TYPE: Dict[str, Any]

create_local_mcp_server

create_local_mcp_server(name: str, script_path: Union[str, Path], interpreter: str = 'python', **kwargs) -> MCPServerConfig

Create configuration for local stdio MCP server.

Source code in packages/ai-parrot/src/parrot/mcp/integration.py

def create_local_mcp_server(
    name: str,
    script_path: Union[str, Path],
    interpreter: str = "python",
    **kwargs
) -> MCPServerConfig:
    """Create configuration for local stdio MCP server."""
    script_path = Path(script_path)
    if not script_path.exists():
        raise FileNotFoundError(f"MCP server script not found: {script_path}")

    return MCPServerConfig(
        name=name,
        command=interpreter,
        args=[str(script_path)],
        transport="stdio",
        **kwargs
    )

create_http_mcp_server

create_http_mcp_server(name: str, url: str, auth_type: Optional[str] = None, auth_config: Optional[Dict[str, Any]] = None, headers: Optional[Dict[str, str]] = None, **kwargs) -> MCPServerConfig

Create configuration for HTTP MCP server.

Source code in packages/ai-parrot/src/parrot/mcp/integration.py

def create_http_mcp_server(
    name: str,
    url: str,
    auth_type: Optional[str] = None,
    auth_config: Optional[Dict[str, Any]] = None,
    headers: Optional[Dict[str, str]] = None,
    **kwargs
) -> MCPServerConfig:
    """Create configuration for HTTP MCP server."""
    return MCPServerConfig(
        name=name,
        url=url,
        transport="http",
        auth_type=auth_type,
        auth_config=auth_config or {},
        headers=headers or {},
        **kwargs
    )

create_api_key_mcp_server

create_api_key_mcp_server(name: str, url: str, api_key: str, header_name: str = 'X-API-Key', use_bearer_prefix: bool = False, **kwargs) -> MCPServerConfig

Create configuration for API key authenticated MCP server.

PARAMETER	DESCRIPTION
name	Unique name for the MCP server TYPE: str
url	Base URL of the MCP server TYPE: str
api_key	API key for authentication TYPE: str
header_name	Header name for the API key (default: "X-API-Key") TYPE: str DEFAULT: 'X-API-Key'
use_bearer_prefix	If True, prepend "Bearer " to the API key value (default: False) TYPE: bool DEFAULT: False
**kwargs	Additional MCPServerConfig parameters DEFAULT: {}

RETURNS	DESCRIPTION
MCPClientConfig	MCPServerConfig instance

Source code in packages/ai-parrot/src/parrot/mcp/integration.py

def create_api_key_mcp_server(
    name: str,
    url: str,
    api_key: str,
    header_name: str = "X-API-Key",
    use_bearer_prefix: bool = False,
    **kwargs
) -> MCPServerConfig:
    """Create configuration for API key authenticated MCP server.

    Args:
        name: Unique name for the MCP server
        url: Base URL of the MCP server
        api_key: API key for authentication
        header_name: Header name for the API key (default: "X-API-Key")
        use_bearer_prefix: If True, prepend "Bearer " to the API key value (default: False)
        **kwargs: Additional MCPServerConfig parameters

    Returns:
        MCPServerConfig instance
    """
    return create_http_mcp_server(
        name=name,
        url=url,
        auth_type="api_key",
        auth_config={
            "api_key": api_key,
            "header_name": header_name,
            "use_bearer_prefix": use_bearer_prefix
        },
        **kwargs
    )

create_netsuite_m2m_mcp_server

create_netsuite_m2m_mcp_server(*, account_id: str, client_id: str, certificate_id: str, private_key_path: str, name: str = 'netsuite', token_store: Optional[TokenStore] = None, headers: Optional[Dict[str, Any]] = None) -> MCPServerConfig

Create a NetSuite MCP server using OAuth2 Client Credentials (M2M) with certificate.

This is the machine-to-machine variant — no browser login required. Authentication uses a JWT client_assertion signed with the private key whose matching X.509 certificate was uploaded to the NetSuite Integration Record.

PARAMETER	DESCRIPTION
account_id	NetSuite account ID (e.g. "4984231"). TYPE: str
client_id	OAuth2 client ID from the NetSuite integration record. TYPE: str
certificate_id	Certificate ID shown in NetSuite after uploading the public certificate (Mapping Key field). TYPE: str
private_key_path	Path to a PEM-encoded RSA private key file. TYPE: str
name	Server name (default "netsuite"). TYPE: str DEFAULT: 'netsuite'
token_store	Optional :class:~parrot.mcp.oauth.TokenStore. TYPE: Optional[TokenStore] DEFAULT: None
headers	Extra HTTP headers. TYPE: Optional[Dict[str, Any]] DEFAULT: None

RETURNS	DESCRIPTION
MCPClientConfig	class:MCPServerConfig configured for NetSuite M2M.

Example

cfg = create_netsuite_m2m_mcp_server( ... account_id="4984231", ... client_id="abc...", ... certificate_id="XYZ123", ... private_key_path="/path/to/private.pem", ... )

Source code in packages/ai-parrot/src/parrot/mcp/integration.py

def create_netsuite_m2m_mcp_server(
    *,
    account_id: str,
    client_id: str,
    certificate_id: str,
    private_key_path: str,
    name: str = "netsuite",
    token_store: Optional[TokenStore] = None,
    headers: Optional[Dict[str, Any]] = None,
) -> MCPServerConfig:
    """Create a NetSuite MCP server using OAuth2 Client Credentials (M2M) with certificate.

    This is the machine-to-machine variant — no browser login required.
    Authentication uses a JWT ``client_assertion`` signed with the private key
    whose matching X.509 certificate was uploaded to the NetSuite Integration Record.

    Args:
        account_id: NetSuite account ID (e.g. ``"4984231"``).
        client_id: OAuth2 client ID from the NetSuite integration record.
        certificate_id: Certificate ID shown in NetSuite after uploading the
            public certificate (Mapping Key field).
        private_key_path: Path to a PEM-encoded RSA private key file.
        name: Server name (default ``"netsuite"``).
        token_store: Optional :class:`~parrot.mcp.oauth.TokenStore`.
        headers: Extra HTTP headers.

    Returns:
        :class:`MCPServerConfig` configured for NetSuite M2M.

    Example:
        >>> cfg = create_netsuite_m2m_mcp_server(
        ...     account_id="4984231",
        ...     client_id="abc...",
        ...     certificate_id="XYZ123",
        ...     private_key_path="/path/to/private.pem",
        ... )
    """
    from .oauth import NetSuiteM2MAuth

    url = NETSUITE_MCP_URL.format(account_id=account_id)
    token_url = NETSUITE_TOKEN_URL.format(account_id=account_id)

    m2m = NetSuiteM2MAuth(
        client_id=client_id,
        certificate_id=certificate_id,
        private_key_path=private_key_path,
        account_id=account_id,
        token_url=token_url,
        scopes=NETSUITE_SCOPES,
        token_store=token_store,
    )

    cfg = MCPServerConfig(
        name=name,
        transport="http",
        url=url,
        headers=headers or {"Content-Type": "application/json"},
        auth_type="oauth",
        auth_config={
            "grant_type": "client_credentials",
            "token_url": token_url,
            "scopes": NETSUITE_SCOPES,
            "client_id": client_id,
            "certificate_id": certificate_id,
        },
        token_supplier=m2m.token_supplier,
    )
    cfg._ensure_oauth_token = m2m.ensure_token
    return cfg

retry_on_errors

retry_on_errors(max_retries: int = 3, base_wait: float = 2.0) -> Callable[[F], F]

Decorator for automatic retry on transient errors with exponential backoff.

PARAMETER	DESCRIPTION
max_retries	Maximum number of retry attempts (default: 3) TYPE: int DEFAULT: 3
base_wait	Base wait time in seconds for exponential backoff (default: 2.0) TYPE: float DEFAULT: 2.0

RETURNS	DESCRIPTION
Callable[[F], F]	Decorated async function with retry logic

Example

@retry_on_errors(max_retries=3) ... async def get_tools(self): ... # This will auto-retry on TransientMCPError ... return await self._session.list_tools()

Source code in packages/ai-parrot/src/parrot/mcp/context.py

def retry_on_errors(max_retries: int = 3, base_wait: float = 2.0) -> Callable[[F], F]:
    """Decorator for automatic retry on transient errors with exponential backoff.

    Args:
        max_retries: Maximum number of retry attempts (default: 3)
        base_wait: Base wait time in seconds for exponential backoff (default: 2.0)

    Returns:
        Decorated async function with retry logic

    Example:
        >>> @retry_on_errors(max_retries=3)
        ... async def get_tools(self):
        ...     # This will auto-retry on TransientMCPError
        ...     return await self._session.list_tools()
    """
    def decorator(func: F) -> F:
        @wraps(func)
        async def wrapper(*args, **kwargs):
            logger = logging.getLogger("MCPRetry")
            last_error = None

            for attempt in range(max_retries):
                try:
                    return await func(*args, **kwargs)
                except TransientMCPError as e:
                    last_error = e
                    if attempt == max_retries - 1:
                        raise

                    wait_time = base_wait ** attempt
                    logger.warning(
                        f"Transient error on attempt {attempt + 1}/{max_retries}, "
                        f"retrying in {wait_time:.1f}s: {e}"
                    )
                    await asyncio.sleep(wait_time)

            # Should not reach here, but just in case
            if last_error:
                raise last_error

        return wrapper  # type: ignore
    return decorator

get_factory_map

get_factory_map() -> Dict[str, Any]

Return the dispatch map from registry slug to create_* factory function.

Deferred import to avoid circular dependencies (the factory functions live in parrot.mcp.integration which may import from this module).

Source code in packages/ai-parrot/src/parrot/mcp/registry.py

def get_factory_map() -> Dict[str, Any]:
    """Return the dispatch map from registry slug to ``create_*`` factory function.

    Deferred import to avoid circular dependencies (the factory functions live
    in ``parrot.mcp.integration`` which may import from this module).
    """
    from parrot.mcp.integration import (
        create_alphavantage_mcp_server,
        create_chrome_devtools_mcp_server,
        create_fireflies_mcp_server,
        create_google_maps_mcp_server,
        create_netsuite_mcp_server,
        create_perplexity_mcp_server,
        create_quic_mcp_server,
        create_websocket_mcp_server,
    )

    return {
        "alphavantage": create_alphavantage_mcp_server,
        "chrome-devtools": create_chrome_devtools_mcp_server,
        "fireflies": create_fireflies_mcp_server,
        "google-maps": create_google_maps_mcp_server,
        "netsuite": create_netsuite_mcp_server,
        "perplexity": create_perplexity_mcp_server,
        "quic": create_quic_mcp_server,
        "websocket": create_websocket_mcp_server,
    }