Library Implementation Principles

Core Principle: Generalizability

When solving specific issues in the penguiflow library, implementations MUST be generalizable. Never hardcode solutions for specific use cases.

Anti-Pattern Examples

Bad: Hardcoding Transport Type

# WRONG - Forces StreamableHTTP for all URL-based connections
if connection.startswith(("http://", "https://")):
    transport = StreamableHttpTransport(url=connection, headers=auth_headers)

Bad: Assuming Auth Mechanism

# WRONG - Only handles one auth type at connection time
if self.config.auth_type == AuthType.BEARER:
    # handle bearer
# Missing: API_KEY, COOKIE, OAUTH2_USER, custom auth

Correct Patterns

1. Transport Detection

FastMCP supports multiple transports - let it detect or make it configurable:

# GOOD - Configurable MCP transport mode (in config.py)
class McpTransportMode(str, Enum):
    """MCP transport mode for URL-based connections."""
    AUTO = "auto"  # Let FastMCP auto-detect (default when no auth headers)
    SSE = "sse"  # Force Server-Sent Events transport (legacy)
    STREAMABLE_HTTP = "streamable_http"  # Force Streamable HTTP (modern)

class ExternalToolConfig(BaseModel):
    # ...
    mcp_transport_mode: McpTransportMode = Field(
        default=McpTransportMode.AUTO,
        description="For MCP over HTTP: auto-detect, sse, or streamable_http",
    )

# GOOD - In ToolNode._resolve_mcp_url_transport()
def _resolve_mcp_url_transport(self, connection, auth_headers, sse_cls, streamable_cls):
    mode = self.config.mcp_transport_mode

    # Explicit SSE mode
    if mode == McpTransportMode.SSE:
        return sse_cls(url=connection, headers=auth_headers or {})

    # Explicit StreamableHTTP mode
    if mode == McpTransportMode.STREAMABLE_HTTP:
        return streamable_cls(url=connection, headers=auth_headers or {})

    # AUTO mode without auth - let FastMCP auto-detect
    if not auth_headers:
        return connection

    # AUTO mode with auth - detect from URL pattern
    if "/sse" in connection.lower():
        return sse_cls(url=connection, headers=auth_headers)

    return streamable_cls(url=connection, headers=auth_headers)

2. Auth Type Extensibility

Support all auth types uniformly:

# GOOD - Extensible auth resolution
class AuthType(str, Enum):
    NONE = "none"
    API_KEY = "api_key"
    BEARER = "bearer"
    COOKIE = "cookie"
    OAUTH2_USER = "oauth2_user"
    CUSTOM = "custom"  # For custom auth handlers

def _get_auth_headers(self) -> dict[str, str]:
    """Get auth headers based on config - supports all auth types."""
    handlers = {
        AuthType.NONE: lambda: {},
        AuthType.API_KEY: self._get_api_key_headers,
        AuthType.BEARER: self._get_bearer_headers,
        AuthType.COOKIE: self._get_cookie_headers,
        # OAuth2 is async and handled separately
    }
    handler = handlers.get(self.config.auth_type)
    return handler() if handler else {}

3. Configuration Over Code

Make behavior configurable, not hardcoded:

# GOOD - Configuration-driven behavior (Pydantic model in config.py)
class ExternalToolConfig(BaseModel):
    name: str
    transport: TransportType = TransportType.MCP
    mcp_transport_mode: McpTransportMode = McpTransportMode.AUTO
    auth_type: AuthType = AuthType.NONE
    auth_config: dict[str, Any] = Field(default_factory=dict)
    # ... other config

MCP Transport Guidelines

FastMCP Auto-Detection

FastMCP can auto-detect transport when given a URL:

• /mcp

  endpoints → typically StreamableHTTP

• /sse

  endpoints → typically SSE

But some servers don't follow this convention.

When to Force Transport

from penguiflow.tools.config import ExternalToolConfig, McpTransportMode, TransportType

# SSE endpoints (legacy)
ExternalToolConfig(
    name="legacy-server",
    transport=TransportType.MCP,
    connection="https://server.com/sse",
    mcp_transport_mode=McpTransportMode.SSE,
)

# StreamableHTTP endpoints (modern)
ExternalToolConfig(
    name="modern-server",
    transport=TransportType.MCP,
    connection="https://server.com/mcp",
    mcp_transport_mode=McpTransportMode.STREAMABLE_HTTP,
)

# Let FastMCP decide (default)
ExternalToolConfig(
    name="auto-detect",
    transport=TransportType.MCP,
    connection="https://server.com/mcp",
    mcp_transport_mode=McpTransportMode.AUTO,
)

Checklist Before Implementation

1. Does this work for ALL auth types, not just the one I'm testing?
2. Does this work for ALL transport types (SSE, StreamableHTTP, STDIO)?
3. Is behavior configurable via ExternalToolConfig?
4. Are there sensible defaults that don't break existing usage?
5. Is error handling comprehensive?
6. Does it follow existing patterns in the codebase?

Testing Requirements

When adding new features:

1. Test with multiple auth types
2. Test with multiple transport types
3. Test with edge cases (missing config, invalid values)
4. Add unit tests AND integration tests

DNA Pattern References

• See

  penguiflow/planner/

  for ReactPlanner patterns
• See

  penguiflow/tools/config.py

  for configuration patterns
• See

  test_generation/*/src/*/external_tools.py

  for usage patterns