[go: up one dir, main page]
Writing your own provider

Python Custom Provider

Learn how to create custom providers for any AI framework in Python

This guide shows how to create custom providers for the Composio Python SDK. Custom providers enable integration with different AI frameworks and platforms.

Provider architecture

The Composio SDK uses a provider architecture to adapt tools for different AI frameworks. The provider handles:

  1. Tool format transformation: Converting Composio tools into formats compatible with specific AI platforms
  2. Platform-specific integration: Providing helper methods for seamless integration
Types of providers

There are two types of providers:

  1. Non-agentic providers: Transform tools for platforms that don’t have their own agency (e.g., OpenAI, Anthropic)
  2. Agentic providers: Transform tools for platforms that have their own agency (e.g., LangChain, CrewAI)
Provider class hierarchy 
BaseProvider (Abstract)
├── NonAgenticProvider (Abstract)
│   └── OpenAIProvider (Concrete)
│   └── AnthropicProvider (Concrete)
│   └── [Your Custom Non-Agentic Provider] (Concrete)
└── AgenticProvider (Abstract)
    └── LangchainProvider (Concrete)
    └── [Your Custom Agentic Provider] (Concrete)
Quick start

The fastest way to create a new provider is using the provider scaffolding script:

$# Create a non-agentic provider
>make create-provider name=myprovider
>
># Create an agentic provider
>make create-provider name=myagent agentic=true
>
># Create provider with custom output directory
>make create-provider name=myprovider output=/path/to/custom/dir
>
># Combine options
>make create-provider name=myagent agentic=true output=/my/custom/path

This will create a new provider in the specified directory (default: python/providers/<provider-name>/) with:

  • Complete package structure with pyproject.toml
  • Provider implementation template
  • Demo script
  • README with usage examples
  • Type annotations and proper inheritance

The scaffolding script creates a fully functional provider template. You just need to implement the tool transformation logic specific to your platform. You can maintain your provider implementation in your own repository.

Generated structure

The create-provider script generates the following structure:

python/providers/<provider-name>/
├── README.md                    # Documentation and usage examples
├── pyproject.toml              # Project configuration
├── setup.py                    # Setup script for pip compatibility
├── <provider-name>_demo.py     # Demo script showing usage
└── composio_<provider-name>/   # Package directory
    ├── __init__.py             # Package initialization
    ├── provider.py             # Provider implementation
    └── py.typed                # PEP 561 type marker

After generation, you can:

  1. Navigate to the provider directory: cd python/providers/<provider-name>
  2. Install in development mode: uv pip install -e .
  3. Implement your provider logic in composio_<provider-name>/provider.py
  4. Test with the demo script: python <provider-name>_demo.py
Creating a non-agentic provider

Non-agentic providers inherit from the NonAgenticProvider abstract class:

1from typing import List, Optional, Sequence, TypeAlias
2from composio.core.provider import NonAgenticProvider
3from composio.types import Tool, Modifiers, ToolExecutionResponse
4
5# Define your tool format
6class MyAITool:
7    def __init__(self, name: str, description: str, parameters: dict):
8        self.name = name
9        self.description = description
10        self.parameters = parameters
11
12# Define your tool collection format
13MyAIToolCollection: TypeAlias = List[MyAITool]
14
15# Create your provider
16class MyAIProvider(NonAgenticProvider[MyAITool, MyAIToolCollection], name="my-ai-platform"):
17    """Custom provider for My AI Platform"""
18    
19    def wrap_tool(self, tool: Tool) -> MyAITool:
20        """Transform a single tool to platform format"""
21        return MyAITool(
22            name=tool.slug,
23            description=tool.description or "",
24            parameters={
25                "type": "object",
26                "properties": tool.input_parameters.get("properties", {}),
27                "required": tool.input_parameters.get("required", [])
28            }
29        )
30    
31    def wrap_tools(self, tools: Sequence[Tool]) -> MyAIToolCollection:
32        """Transform a collection of tools"""
33        return [self.wrap_tool(tool) for tool in tools]
34    
35    # Optional: Custom helper methods for your AI platform
36    def execute_my_ai_tool_call(
37        self,
38        user_id: str,
39        tool_call: dict,
40        modifiers: Optional[Modifiers] = None
41    ) -> ToolExecutionResponse:
42        """Execute a tool call in your platform's format
43        
44        Example usage:
45        result = my_provider.execute_my_ai_tool_call(
46            user_id="default",
47            tool_call={"name": "GITHUB_STAR_REPO", "arguments": {"owner": "composiohq", "repo": "composio"}}
48        )
49        """
50        # Use the built-in execute_tool method
51        return self.execute_tool(
52            slug=tool_call["name"],
53            arguments=tool_call["arguments"],
54            modifiers=modifiers,
55            user_id=user_id
56        )
Creating an agentic provider

Agentic providers inherit from the AgenticProvider abstract class:

1from typing import Callable, Dict, List, Sequence
2from composio.core.provider import AgenticProvider, AgenticProviderExecuteFn
3from composio.types import Tool
4from my_provider import AgentTool
5
6# Import the Tool/Function class that represents a callable tool for your framework
7# Optionally define your custom tool format below
8# class AgentTool:
9#     def __init__(self, name: str, description: str, execute: Callable, schema: dict):
10#         self.name = name
11#         self.description = description
12#         self.execute = execute
13#         self.schema = schema
14
15# Define your tool collection format (typically a List)
16AgentToolCollection: TypeAlias = List[AgentTool]
17
18# Create your provider
19class MyAgentProvider(AgenticProvider[AgentTool, List[AgentTool]], name="my-agent-platform"):
20    """Custom provider for My Agent Platform"""
21    
22    def wrap_tool(self, tool: Tool, execute_tool: AgenticProviderExecuteFn) -> AgentTool:
23        """Transform a single tool with execute function"""
24        def execute_wrapper(**kwargs) -> Dict:
25            result = execute_tool(tool.slug, kwargs)
26            if not result.get("successful", False):
27                raise Exception(result.get("error", "Tool execution failed"))
28            return result.get("data", {})
29        
30        return AgentTool(
31            name=tool.slug,
32            description=tool.description or "",
33            execute=execute_wrapper,
34            schema=tool.input_parameters
35        )
36    
37    def wrap_tools(
38        self, 
39        tools: Sequence[Tool], 
40        execute_tool: AgenticProviderExecuteFn
41    ) -> AgentToolCollection:
42        """Transform a collection of tools with execute function"""
43        return [self.wrap_tool(tool, execute_tool) for tool in tools]
Using your custom provider

After creating your provider, use it with the Composio SDK:

Non-agentic provider example 
1from composio import Composio
2from composio_myai import MyAIProvider
3from myai import MyAIClient  # Your AI platform's client
4
5# Initialize tools
6myai_client = MyAIClient()
7composio = Composio(provider=MyAIProvider())
8
9# Define task
10task = "Star a repo composiohq/composio on GitHub"
11
12# Get GitHub tools that are pre-configured
13tools = composio.tools.get(user_id="default", toolkits=["GITHUB"])
14
15# Get response from your AI platform (example)
16response = myai_client.chat.completions.create(
17    model="your-model",
18    tools=tools,  # These are in your platform's format
19    messages=[
20        {"role": "system", "content": "You are a helpful assistant."},
21        {"role": "user", "content": task},
22    ],
23)
24print(response)
25
26# Execute the function calls
27result = composio.provider.handle_tool_calls(response=response, user_id="default")
28print(result)
Agentic provider example 
1import asyncio
2from agents import Agent, Runner
3from composio_myagent import MyAgentProvider
4from composio import Composio
5
6# Initialize Composio toolset
7composio = Composio(provider=MyAgentProvider())
8
9# Get all the tools
10tools = composio.tools.get(
11    user_id="default",
12    tools=["GITHUB_STAR_A_REPOSITORY_FOR_THE_AUTHENTICATED_USER"],
13)
14
15# Create an agent with the tools
16agent = Agent(
17    name="GitHub Agent",
18    instructions="You are a helpful assistant that helps users with GitHub tasks.",
19    tools=tools,
20)
21
22# Run the agent
23async def main():
24    result = await Runner.run(
25        starting_agent=agent,
26        input=(
27            "Star the repository composiohq/composio on GitHub. If done "
28            "successfully, respond with 'Action executed successfully'"
29        ),
30    )
31    print(result.final_output)
32
33asyncio.run(main())
Example: Anthropic Claude provider

Here’s a complete example for Anthropic’s Claude:

1import typing as t
2
3from anthropic.types.beta.beta_tool_use_block import BetaToolUseBlock
4from anthropic.types.message import Message as ToolsBetaMessage
5from anthropic.types.tool_param import ToolParam
6from anthropic.types.tool_use_block import ToolUseBlock
7
8from composio.core.provider import NonAgenticProvider
9from composio.types import Modifiers, Tool, ToolExecutionResponse
10
11
12class AnthropicProvider(
13    NonAgenticProvider[ToolParam, list[ToolParam]],
14    name="anthropic",
15):
16    """Composio toolset for Anthropic Claude platform."""
17
18    def wrap_tool(self, tool: Tool) -> ToolParam:
19        return ToolParam(
20            input_schema=tool.input_parameters,
21            name=tool.slug,
22            description=tool.description,
23        )
24
25    def wrap_tools(self, tools: t.Sequence[Tool]) -> list[ToolParam]:
26        return [self.wrap_tool(tool) for tool in tools]
27
28    def execute_tool_call(
29        self,
30        user_id: str,
31        tool_call: ToolUseBlock,
32        modifiers: t.Optional[Modifiers] = None,
33    ) -> ToolExecutionResponse:
34        """Execute a tool call.
35        
36        :param user_id: User ID to use for executing function calls.
37        :param tool_call: Tool call metadata.
38        :param modifiers: Modifiers to use for executing function calls.
39        :return: Object containing output data from the tool call.
40        """
41        return self.execute_tool(
42            slug=tool_call.name,
43            arguments=t.cast(t.Dict, tool_call.input),
44            modifiers=modifiers,
45            user_id=user_id,
46        )
47
48    def handle_tool_calls(
49        self,
50        user_id: str,
51        response: t.Union[dict, ToolsBetaMessage],
52        modifiers: t.Optional[Modifiers] = None,
53    ) -> t.List[ToolExecutionResponse]:
54        """Handle tool calls from Anthropic Claude chat completion object.
55        
56        :param response: Chat completion object from
57            `anthropic.Anthropic.beta.tools.messages.create` function call.
58        :param user_id: User ID to use for executing function calls.
59        :param modifiers: Modifiers to use for executing function calls.
60        :return: A list of output objects from the tool calls.
61        """
62        if isinstance(response, dict):
63            response = ToolsBetaMessage(**response)
64
65        outputs = []
66        for content in response.content:
67            if isinstance(content, (ToolUseBlock, BetaToolUseBlock)):
68                outputs.append(
69                    self.execute_tool_call(
70                        user_id=user_id,
71                        tool_call=content,
72                        modifiers=modifiers,
73                    )
74                )
75        return outputs
Example: OpenAI Agents provider

Here’s an example for OpenAI Agents framework:

1import asyncio
2import json
3import typing as t
4
5import pydantic
6from agents import FunctionTool
7
8from composio.core.provider import AgenticProvider, AgenticProviderExecuteFn
9from composio.types import Tool
10from composio.utils.pydantic import parse_pydantic_error
11
12
13class OpenAIAgentsProvider(
14    AgenticProvider[FunctionTool, list[FunctionTool]],
15    name="openai_agents",
16):
17    """Provider for OpenAI Agents framework"""
18    
19    def wrap_tool(
20        self,
21        tool: Tool,
22        execute_tool: AgenticProviderExecuteFn,
23    ) -> FunctionTool:
24        """Wrap a tool as a FunctionTool"""
25        
26        # Create an async wrapper for the tool execution
27        async def execute_tool_wrapper(_ctx, payload):
28            """Execute Composio action with the given arguments"""
29            try:
30                return json.dumps(
31                    obj=(
32                        await asyncio.to_thread(  # Running in thread since execute_tool is not async
33                            execute_tool,
34                            slug=tool.slug,
35                            arguments=json.loads(payload) if payload else {},
36                        )
37                    )
38                )
39            except pydantic.ValidationError as e:
40                return json.dumps({
41                    "successful": False,
42                    "error": parse_pydantic_error(e),
43                    "data": None,
44                })
45            except Exception as e:
46                return json.dumps({
47                    "successful": False,
48                    "error": str(e),
49                    "data": None,
50                })
51        
52        # Ensure the schema has additionalProperties set to false
53        modified_schema = tool.input_parameters.copy()
54        modified_schema["additionalProperties"] = False
55        
56        # Create a FunctionTool with the appropriate schema
57        return FunctionTool(
58            name=tool.slug,
59            description=tool.description,
60            params_json_schema=modified_schema,
61            on_invoke_tool=execute_tool_wrapper,
62            strict_json_schema=False,  # Composio tools already have optimal schemas
63        )
64    
65    def wrap_tools(
66        self,
67        tools: t.Sequence[Tool],
68        execute_tool: AgenticProviderExecuteFn,
69    ) -> list[FunctionTool]:
70        """Wrap multiple tools for OpenAI Agents"""
71        return [self.wrap_tool(tool, execute_tool) for tool in tools]
Best practices
  1. Keep providers focused: Each provider should integrate with one specific platform
  2. Handle errors gracefully: Catch and transform errors from tool execution
  3. Follow platform conventions: Adopt naming and structural conventions of the target platform
  4. Use type annotations: Leverage Python’s typing system for better IDE support and documentation
  5. Cache transformed tools: Store transformed tools when appropriate to improve performance
  6. Add helper methods: Provide convenient methods for common platform-specific operations
  7. Document your provider: Include docstrings and usage examples
  8. Set meaningful provider names: Use the name parameter for telemetry and debugging