ayn

ayn - Agents You Need

A meta-framework for discovering, installing, and integrating reusable AI agents. Acts as the ‘pip + Docker Hub + Postman’ for AI agents.

Core Features: 1. Search & Discovery: Find agents across GitHub, HuggingFace, awesome lists 2. Standard Controller Interface: Parametrizable agent controllers with action methods 3. HTTP/REST Export: FastAPI and LangServe wrappers 4. MCP Support: Model Context Protocol server integration 5. Custom Actions: ChatGPT and Claude action generation

Architecture: - Search layer: Unified registry across multiple sources - Controller protocol: Standard interface for invoking agents - Export layer: HTTP/REST service generation - MCP layer: Tools, resources, and prompts - Actions: OpenAPI specs for ChatGPT/Claude

Example

>>> from ayn import AgentRegistry, BaseAgentController
>>> registry = AgentRegistry()
>>> results = registry.search('data preparation agents')
>>> agent = BaseAgentController.from_metadata(results[0])
>>> result = agent.invoke(input_data)

class ayn.AgentController(*args, **kwargs)[source]

Standard protocol for agent controllers

Controllers provide a uniform interface for invoking agents regardless of their underlying framework.

Example

>>> class MyController:
...     def __init__(self, config):
...         self.config = config
...     def invoke(self, input_data, **kwargs):
...         return f"Processed: {input_data}"
>>> controller = MyController({'model': 'gpt-4'})
>>> controller.invoke('test')
'Processed: test'

async ainvoke(input_data: Any, **kwargs) → Any[source]: Invoke the agent asynchronously

invoke(input_data: Any, **kwargs) → Any[source]

Invoke the agent synchronously

Parameters:

input_data – Input data for the agent
**kwargs – Additional parameters

Returns:

Agent output

stream(input_data: Any, **kwargs) → Iterable[Any][source]: Stream agent outputs

class ayn.AgentFramework(value)[source]: Supported agent frameworks

class ayn.AgentMetadata(name: str, description: str, framework: AgentFramework, source: str = '', tags: List[str] = <factory>, modality: AgentModality = AgentModality.TEXT, version: str = '0.1.0', author: str = '', license: str = '', dependencies: List[str] = <factory>, capabilities: List[str] = <factory>)[source]

Metadata for an agent

Example

>>> meta = AgentMetadata(
...     name="data-prep-agent",
...     description="Prepares data for visualization",
...     framework=AgentFramework.CREWAI
... )
>>> meta.name
'data-prep-agent'

classmethod from_dict(data: dict) → AgentMetadata[source]: Create from dictionary

to_dict() → dict[source]

Convert to dictionary

Example

>>> meta = AgentMetadata(name="test", description="test agent", framework=AgentFramework.CREWAI)
>>> d = meta.to_dict()
>>> d['name']
'test'

class ayn.AgentModality(value)[source]: Agent modalities

class ayn.AgentRegistry(searchers: List[AgentSearcher] | None = None, cache_dir: str | None = None)[source]

Unified registry for searching and managing agents

Aggregates results from multiple sources and provides a dict-like interface.

Example

>>> registry = AgentRegistry()
>>> # Add a local agent
>>> from ayn.core import AgentMetadata, AgentFramework
>>> meta = AgentMetadata(name="test", description="test", framework=AgentFramework.CUSTOM)
>>> registry['test'] = meta
>>> registry['test'].name
'test'
>>> 'test' in registry
True

search(query: str, *, limit: int = 20, source: str | None = None, **filters) → List[AgentMetadata][source]

Search across all sources

Parameters:

query – Search query
limit – Max results
source – Specific source to search (github, huggingface, awesome)
**filters – Additional filters

Returns:

List of agent metadata

Example

>>> registry = AgentRegistry(searchers=[])
>>> # Add some test data
>>> from ayn.core import AgentMetadata, AgentFramework
>>> registry['agent1'] = AgentMetadata(
...     name="agent1", description="test agent",
...     framework=AgentFramework.CREWAI, tags=['data']
... )
>>> results = registry.search('agent')
>>> len(results) >= 1
True

class ayn.AgentSearcher(*args, **kwargs)[source]

Protocol for agent search implementations

search(query: str, *, limit: int = 20, **filters) → Iterable[AgentMetadata][source]

Search for agents

Parameters:

query – Search query string
limit – Maximum results to return
**filters – Additional filters (framework, tags, etc.)

Returns:

Iterable of agent metadata

class ayn.AutoGenController(metadata: AgentMetadata, config: ControllerConfig)[source]

Controller for AutoGen agents

TODO: Implement full AutoGen integration See AYN_IMPLEMENTATION_ROADMAP.md for details

class ayn.AwesomeListSearcher[source]

Parse and search awesome lists for agents

Example

>>> searcher = AwesomeListSearcher()
>>> results = searcher.search('autogen')

search(query: str, *, limit: int = 20, **filters) → Iterable[AgentMetadata][source]: Search awesome lists

class ayn.BaseAgentController(config: ControllerConfig)[source]

Base class for agent controllers

Provides common functionality and enforces the controller protocol.

Example

>>> class SimpleController(BaseAgentController):
...     def invoke(self, input_data, **kwargs):
...         return f"Result: {input_data}"
>>> controller = SimpleController(ControllerConfig())
>>> controller.invoke('test')
'Result: test'

async ainvoke(input_data: Any, **kwargs) → Any[source]: Default async implementation wraps synchronous invoke

classmethod from_metadata(metadata: AgentMetadata, config: ControllerConfig | None = None) → BaseAgentController[source]: Factory method to create controller from metadata

abstractmethod invoke(input_data: Any, **kwargs) → Any[source]: Invoke the agent - must be implemented by subclasses

stream(input_data: Any, **kwargs) → Iterable[Any][source]: Default streaming yields single result

class ayn.ControllerConfig(model: str = 'gpt-4', temperature: float = 0.7, max_tokens: int = 1000, timeout: int = 60, retry_count: int = 3, api_key: str | None = None, additional_params: dict = <factory>)[source]

Configuration for agent controllers

Example

>>> config = ControllerConfig(
...     model="gpt-4",
...     temperature=0.7,
...     max_tokens=1000
... )
>>> config.model
'gpt-4'

to_dict() → dict[source]: Convert to dictionary

class ayn.CrewAIController(metadata: AgentMetadata, config: ControllerConfig)[source]

Controller for CrewAI agents

TODO: Implement full CrewAI integration See AYN_IMPLEMENTATION_ROADMAP.md for details

class ayn.GenericController(metadata: AgentMetadata, config: ControllerConfig)[source]

Generic controller for custom agents

Example

>>> from ayn.core import AgentMetadata, AgentFramework, ControllerConfig
>>> meta = AgentMetadata(name="test", description="test", framework=AgentFramework.CUSTOM)
>>> controller = GenericController(meta, ControllerConfig())
>>> result = controller.invoke({'input': 'test'})

invoke(input_data: Any, **kwargs) → Any[source]: Generic invoke - subclasses override for specific behavior

class ayn.GitHubAgentSearcher(github_token: str | None = None)[source]

Search for agents in GitHub repositories

Example

>>> searcher = GitHubAgentSearcher()
>>> results = searcher.search('crewai agent')

search(query: str, *, limit: int = 20, **filters) → Iterable[AgentMetadata][source]: Search GitHub for agent repos

class ayn.HuggingFaceAgentSearcher(hf_token: str | None = None)[source]

Search for agents on Hugging Face

Uses the Hugging Face MCP tool if available.

Example

>>> searcher = HuggingFaceAgentSearcher()
>>> results = searcher.search('code generation')

search(query: str, *, limit: int = 20, **filters) → Iterable[AgentMetadata][source]: Search Hugging Face for agents

class ayn.LangChainController(metadata: AgentMetadata, config: ControllerConfig)[source]

Controller for LangChain agents

TODO: Implement full LangChain integration See AYN_IMPLEMENTATION_ROADMAP.md for details

class ayn.MCPPrompt(name: str, description: str, arguments: List[dict] = <factory>)[source]

MCP prompt template

Example

>>> prompt = MCPPrompt(
...     name="summarize",
...     description="Summarize text",
...     arguments=[{"name": "text", "description": "Text to summarize", "required": True}]
... )
>>> prompt.name
'summarize'

to_dict() → dict[source]: Convert to MCP prompt format

class ayn.MCPResource(uri: str, name: str, mimeType: str = 'text/plain', description: str = '')[source]

MCP resource definition

Example

>>> resource = MCPResource(
...     uri="file:///data.csv",
...     name="Dataset",
...     mimeType="text/csv"
... )
>>> resource.name
'Dataset'

to_dict() → dict[source]: Convert to MCP resource format

class ayn.MCPServer(name: str, version: str = '1.0.0', controller: AgentController | None = None)[source]

MCP Server implementation

Exposes agent capabilities via Model Context Protocol.

Example

>>> server = MCPServer(name="test-server")
>>> tool = MCPTool(
...     name="greet",
...     description="Greet user",
...     input_schema={"type": "object", "properties": {}}
... )
>>> server.add_tool(tool)
>>> len(server.tools)
1

add_prompt(prompt: MCPPrompt)[source]: Add a prompt to the server

add_resource(resource: MCPResource)[source]: Add a resource to the server

add_tool(tool: MCPTool)[source]

Add a tool to the server

Example

>>> server = MCPServer("test")
>>> tool = MCPTool("test", "test tool", {})
>>> server.add_tool(tool)
>>> server.tools[0].name
'test'

handle_call_tool(tool_name: str, arguments: dict) → dict[source]: Handle tools/call request

handle_list_prompts() → dict[source]: Handle prompts/list request

handle_list_resources() → dict[source]: Handle resources/list request

handle_list_tools() → dict[source]: Handle tools/list request

to_json_rpc_handler() → Callable[source]

Create a JSON-RPC handler for MCP protocol

Returns:: Handler function that processes JSON-RPC requests

class ayn.MCPTool(name: str, description: str, input_schema: dict, handler: Callable | None = None)[source]

MCP tool definition

Example

>>> tool = MCPTool(
...     name="search",
...     description="Search for information",
...     input_schema={"type": "object", "properties": {"query": {"type": "string"}}}
... )
>>> tool.name
'search'

to_dict() → dict[source]: Convert to MCP tool format

ayn.controller_to_mcp_server(controller: AgentController, metadata: AgentMetadata, *, tools: List[MCPTool] | None = None, resources: List[MCPResource] | None = None, prompts: List[MCPPrompt] | None = None) → MCPServer[source]

Convert an agent controller to MCP server

Parameters:

controller – Agent controller
metadata – Agent metadata
tools – Optional tools to expose
resources – Optional resources to expose
prompts – Optional prompts to expose

Returns:

MCP server instance

Example

>>> from ayn.controllers import GenericController
>>> from ayn.core import AgentMetadata, AgentFramework, ControllerConfig
>>> meta = AgentMetadata(name="test", description="test", framework=AgentFramework.CUSTOM)
>>> controller = GenericController(meta, ControllerConfig())
>>> server = controller_to_mcp_server(controller, meta)
>>> server.name
'test'

ayn.create_agent_from_registry(registry: AgentRegistry, query: str, config: ControllerConfig | None = None) → BaseAgentController[source]

Search registry and create agent controller

Parameters:

registry – Agent registry
query – Search query
config – Optional controller config

Returns:

Agent controller

Example

>>> from ayn import AgentRegistry
>>> from ayn.core import AgentMetadata, AgentFramework
>>> registry = AgentRegistry(searchers=[])
>>> meta = AgentMetadata(name="test-agent", description="test", framework=AgentFramework.CUSTOM)
>>> registry['test-agent'] = meta
>>> controller = create_agent_from_registry(registry, 'test')
>>> isinstance(controller, BaseAgentController)
True

ayn.export_agent_full_stack(controller: AgentController, metadata: AgentMetadata, *, api_host: str = '0.0.0.0', api_port: int = 8000, export_fastapi: bool = True, export_mcp: bool = True, export_chatgpt: bool = True, mcp_tools: List[MCPTool] | None = None) → dict[source]

Export agent with all integrations

Parameters:

controller – Agent controller
metadata – Agent metadata
api_host – API host
api_port – API port
export_fastapi – Whether to export as FastAPI
export_mcp – Whether to export as MCP server
export_chatgpt – Whether to generate ChatGPT action
mcp_tools – Optional MCP tools

Returns:

Dict with export artifacts

Example

>>> from ayn.controllers import GenericController
>>> from ayn.core import AgentMetadata, AgentFramework, ControllerConfig
>>> meta = AgentMetadata(name="test", description="test", framework=AgentFramework.CUSTOM)
>>> controller = GenericController(meta, ControllerConfig())
>>> artifacts = export_agent_full_stack(
...     controller, meta,
...     export_fastapi=False, export_mcp=False, export_chatgpt=True
... )
>>> 'chatgpt_action' in artifacts
True

ayn.export_as_fastapi(controller: AgentController, metadata: AgentMetadata | None = None, host: str = '0.0.0.0', port: int = 8000)[source]

Export agent controller as FastAPI service

Parameters:

controller – Agent controller
metadata – Optional agent metadata
host – Host to bind to
port – Port to bind to

Example

>>> # This would start a server - doctest: +SKIP
>>> controller = GenericController(...)
>>> export_as_fastapi(controller, port=8080)

ayn.export_as_langserve(controller: AgentController, metadata: AgentMetadata | None = None) → Any[source]

Export agent as LangServe compatible service

Parameters:

controller – Agent controller
metadata – Optional agent metadata

Returns:

LangServe app

Example

>>> # Requires langserve - doctest: +SKIP
>>> controller = GenericController(...)
>>> app = export_as_langserve(controller)

ayn.generate_chatgpt_action(controller: AgentController, metadata: AgentMetadata, api_url: str) → dict[source]

Generate ChatGPT custom action configuration

Parameters:

controller – Agent controller
metadata – Agent metadata
api_url – Base URL for the API

Returns:

ChatGPT action configuration

Example

>>> from ayn.controllers import GenericController
>>> from ayn.core import AgentMetadata, AgentFramework, ControllerConfig
>>> meta = AgentMetadata(name="test", description="test", framework=AgentFramework.CUSTOM)
>>> controller = GenericController(meta, ControllerConfig())
>>> action = generate_chatgpt_action(controller, meta, "https://api.example.com")
>>> action['schema']['info']['title']
'test'

ayn.generate_claude_mcp_config(mcp_server: MCPServer, server_url: str, transport: str = 'sse') → dict[source]

Generate Claude Desktop MCP configuration

Parameters:

mcp_server – MCP server instance
server_url – URL where MCP server is hosted
transport – Transport type (sse, stdio)

Returns:

Claude Desktop MCP config

Example

>>> from ayn.mcp import MCPServer
>>> server = MCPServer("test")
>>> config = generate_claude_mcp_config(server, "http://localhost:8000")
>>> config['mcpServers']['test']['url']
'http://localhost:8000/sse'