feat: introduce agent orchestrator to bridge agents and backend server (#20)

Author: vcfgv

python/third_party/TradingAgents/adapter/__main__.py

@@ -10,7 +10,7 @@
 from langgraph.graph import StateGraph, MessagesState, START, END
 from pydantic import BaseModel, Field, field_validator
 from valuecell.core.agent.decorator import create_wrapped_agent
-from valuecell.core.agent.types import BaseAgent
+from valuecell.core.types import BaseAgent
 
 from tradingagents.graph.trading_graph import TradingAgentsGraph
 from tradingagents.default_config import DEFAULT_CONFIG

python/third_party/ai-hedge-fund/adapter/__main__.py

@@ -9,7 +9,7 @@
 from langchain_core.messages import HumanMessage
 from pydantic import BaseModel, Field, field_validator
 from valuecell.core.agent.decorator import create_wrapped_agent
-from valuecell.core.agent.types import BaseAgent
+from valuecell.core.types import BaseAgent
 
 from src.main import create_workflow
 from src.utils.analysts import ANALYST_ORDER

python/valuecell/agents/__init__.py

@@ -9,7 +9,7 @@
 from pathlib import Path
 from typing import List
 
-from valuecell.core.agent.types import BaseAgent
+from valuecell.core.types import BaseAgent
 
 
 def _discover_and_import_agents() -> List[str]:

python/valuecell/agents/hello_world.py

@@ -1,5 +1,5 @@
 from valuecell.core.agent.decorator import serve
-from valuecell.core.agent.types import BaseAgent
+from valuecell.core.types import BaseAgent
 
 
 @serve()
@@ -9,7 +9,7 @@ class HelloWorldAgent(BaseAgent):
     """
 
     async def stream(self, query, session_id, task_id):
-        return {
+        yield {
             "content": f"Hello! You said: {query}",
             "is_task_complete": True,
         }

python/valuecell/agents/sec_13F_agent.py

@@ -7,7 +7,7 @@
 from edgar import Company, set_identity
 from pydantic import BaseModel, Field, field_validator
 
-from valuecell.core.agent.types import BaseAgent
+from valuecell.core.types import BaseAgent
 from valuecell.core.agent.decorator import create_wrapped_agent
 
 # Configure logging

python/valuecell/agents/tests/test_import.py

@@ -1,4 +1,5 @@
 import pytest
+from a2a.types import AgentCard
 from valuecell.core.agent.connect import RemoteConnections
 
 
@@ -10,12 +11,15 @@ async def test_run_hello_world():
         available = connections.list_available_agents()
         assert name in available
 
-        url = await connections.start_agent("HelloWorldAgent")
-        assert isinstance(url, str) and url
+        agent_card = await connections.start_agent("HelloWorldAgent")
+        assert isinstance(agent_card, AgentCard) and agent_card
 
         client = await connections.get_client("HelloWorldAgent")
-        task, event = await client.send_message("Hi there!")
-        assert task is not None
-        assert event is None
+        turns = 0
+        async for task, event in await client.send_message("Hi there!"):
+            assert task is not None
+            assert event is None
+            turns += 1
+        assert turns == 1
     finally:
         await connections.stop_all()

python/valuecell/core/__init__.py

@@ -0,0 +1,49 @@
+# Session management
+from .session import (
+    InMemorySessionStore,
+    Message,
+    Role,
+    Session,
+    SessionManager,
+    SessionStore,
+)
+
+# Task management
+from .task import (
+    InMemoryTaskStore,
+    Task,
+    TaskManager,
+    TaskStatus,
+    TaskStore,
+)
+
+# Type system
+from .types import (
+    UserInput,
+    UserInputMetadata,
+    BaseAgent,
+    StreamResponse,
+    RemoteAgentResponse,
+)
+
+__all__ = [
+    # Session exports
+    "Message",
+    "Role",
+    "Session",
+    "SessionManager",
+    "SessionStore",
+    "InMemorySessionStore",
+    # Task exports
+    "Task",
+    "TaskStatus",
+    "TaskManager",
+    "TaskStore",
+    "InMemoryTaskStore",
+    # Type system exports
+    "UserInput",
+    "UserInputMetadata",
+    "BaseAgent",
+    "StreamResponse",
+    "RemoteAgentResponse",
+]

python/valuecell/core/agent/__init__.py

@@ -0,0 +1,22 @@
+"""Agent module initialization"""
+
+# Core agent functionality
+from .client import AgentClient
+from .connect import RemoteConnections
+from .decorator import serve
+from .registry import AgentRegistry
+
+# Import types from the unified types module
+from ..types import BaseAgent, RemoteAgentResponse, StreamResponse
+
+
+__all__ = [
+    # Core agent exports
+    "AgentClient",
+    "RemoteConnections",
+    "serve",
+    "AgentRegistry",
+    "BaseAgent",
+    "RemoteAgentResponse",
+    "StreamResponse",
+]

python/valuecell/core/agent/client.py

@@ -5,7 +5,7 @@
 from a2a.types import Message, Part, PushNotificationConfig, Role, TextPart
 from valuecell.utils import generate_uuid
 
-from .types import MessageResponse
+from ..types import RemoteAgentResponse
 
 
 class AgentClient:
@@ -48,8 +48,12 @@ async def _setup_client(self):
         self._client = client_factory.create(card)
 
     async def send_message(
-        self, text: str, context_id: str = None, streaming: bool = False
-    ) -> MessageResponse | AsyncIterator[MessageResponse]:
+        self,
+        query: str,
+        context_id: str = None,
+        metadata: dict = None,
+        streaming: bool = False,
+    ) -> AsyncIterator[RemoteAgentResponse]:
         """Send message to Agent.
 
         If `streaming` is True, return an async iterator producing (task, event) pairs.
@@ -59,18 +63,28 @@ async def send_message(
 
         message = Message(
             role=Role.user,
-            parts=[Part(root=TextPart(text=text))],
+            parts=[Part(root=TextPart(text=query))],
             message_id=generate_uuid("msg"),
             context_id=context_id or generate_uuid("ctx"),
+            metadata=metadata if metadata else None,
         )
 
-        generator = self._client.send_message(message)
-        if streaming:
-            return generator
-
-        task, event = await generator.__anext__()
-        await generator.aclose()
-        return task, event
+        source_gen = self._client.send_message(message)
+
+        async def wrapper() -> AsyncIterator[RemoteAgentResponse]:
+            try:
+                if streaming:
+                    async for item in source_gen:
+                        yield item
+                else:
+                    # yield only the first item
+                    item = await source_gen.__anext__()
+                    yield item
+            finally:
+                # ensure underlying generator is closed
+                await source_gen.aclose()
+
+        return wrapper()
 
     async def get_agent_card(self):
         await self._ensure_initialized()