Temporal MCP upstream-session proxy

examples/mcp_agent_server/asyncio/basic_agent_server.py

@@ -188,7 +188,6 @@ async def grade_story(story: str, app_ctx: Optional[AppContext] = None) -> str:
 async def grade_story_async(story: str, app_ctx: Optional[AppContext] = None) -> str:
     """
     Async variant of grade_story that starts a workflow run and returns IDs.
-
     Args:
         story: The student's short story to grade
         app_ctx: Optional MCPApp context for accessing app resources and logging

examples/mcp_agent_server/temporal/basic_agent_server.py

@@ -59,12 +59,20 @@ async def run(
         context = app.context
         context.config.mcp.servers["filesystem"].args.extend([os.getcwd()])
 
+        # Use of the app.logger will forward logs back to the mcp client
+        app_logger = app.logger
+
+        app_logger.info("Starting finder agent")
         async with finder_agent:
             finder_llm = await finder_agent.attach_llm(OpenAIAugmentedLLM)
 
             result = await finder_llm.generate_str(
                 message=input,
             )
+
+            # forwards the log to the caller
+            app_logger.info(f"Finder agent completed with result {result}")
+            # print to the console (for when running locally)
             print(f"Agent result: {result}")
             return WorkflowResult(value=result)
 

examples/mcp_agent_server/temporal/client.py

@@ -1,12 +1,17 @@
 import asyncio
 import json
 import time
-from mcp.types import CallToolResult
 from mcp_agent.app import MCPApp
 from mcp_agent.config import MCPServerSettings
 from mcp_agent.executor.workflow import WorkflowExecution
 from mcp_agent.mcp.gen_client import gen_client
 
+from datetime import timedelta
+from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream
+from mcp import ClientSession
+from mcp_agent.mcp.mcp_agent_client_session import MCPAgentClientSession
+from mcp.types import CallToolResult, LoggingMessageNotificationParams
+
 try:
     from exceptiongroup import ExceptionGroup as _ExceptionGroup  # Python 3.10 backport
 except Exception:  # pragma: no cover
@@ -24,21 +29,45 @@ async def main():
         logger = client_app.logger
         context = client_app.context
 
-        # Connect to the workflow server
-        logger.info("Connecting to workflow server...")
-
-        # Override the server configuration to point to our local script
-        context.server_registry.registry["basic_agent_server"] = MCPServerSettings(
-            name="basic_agent_server",
-            description="Local workflow server running the basic agent example",
-            transport="sse",
-            url="http://0.0.0.0:8000/sse",
-        )
-
         # Connect to the workflow server
         try:
+            logger.info("Connecting to workflow server...")
+
+            # Override the server configuration to point to our local script
+            context.server_registry.registry["basic_agent_server"] = MCPServerSettings(
+                name="basic_agent_server",
+                description="Local workflow server running the basic agent example",
+                transport="sse",
+                url="http://0.0.0.0:8000/sse",
+            )
+
+            # Connect to the workflow server
+            # Define a logging callback to receive server-side log notifications
+            async def on_server_log(params: LoggingMessageNotificationParams) -> None:
+                # Pretty-print server logs locally for demonstration
+                level = params.level.upper()
+                name = params.logger or "server"
+                # params.data can be any JSON-serializable data
+                print(f"[SERVER LOG] [{level}] [{name}] {params.data}")
+
+            # Provide a client session factory that installs our logging callback
+            def make_session(
+                read_stream: MemoryObjectReceiveStream,
+                write_stream: MemoryObjectSendStream,
+                read_timeout_seconds: timedelta | None,
+            ) -> ClientSession:
+                return MCPAgentClientSession(
+                    read_stream=read_stream,
+                    write_stream=write_stream,
+                    read_timeout_seconds=read_timeout_seconds,
+                    logging_callback=on_server_log,
+                )
+
+            # Connect to the workflow server
             async with gen_client(
-                "basic_agent_server", context.server_registry
+                "basic_agent_server",
+                context.server_registry,
+                client_session_factory=make_session,
             ) as server:
                 # Call the BasicAgentWorkflow
                 run_result = await server.call_tool(
@@ -56,6 +85,17 @@ async def main():
                     f"Started BasicAgentWorkflow-run. workflow ID={execution.workflow_id}, run ID={run_id}"
                 )
 
+                get_status_result = await server.call_tool(
+                    "workflows-BasicAgentWorkflow-get_status",
+                    arguments={"run_id": run_id},
+                )
+
+                execution = WorkflowExecution(**json.loads(run_result.content[0].text))
+                run_id = execution.run_id
+                logger.info(
+                    f"Started BasicAgentWorkflow-run. workflow ID={execution.workflow_id}, run ID={run_id}"
+                )
+
                 # Wait for the workflow to complete
                 while True:
                     get_status_result = await server.call_tool(

src/mcp_agent/app.py

@@ -195,12 +195,14 @@ def logger(self):
             try:
                 if self._context is not None:
                     self._logger._bound_context = self._context  # type: ignore[attr-defined]
+
             except Exception:
                 pass
         else:
             # Update the logger's bound context in case upstream_session was set after logger creation
             if self._context and hasattr(self._logger, "_bound_context"):
                 self._logger._bound_context = self._context
+
         return self._logger
 
     async def initialize(self):
@@ -840,6 +842,13 @@ def decorator(target: Callable[..., R]) -> Callable[..., R]:
             )
 
             if task_defn:
+                # prevent trying to decorate an already decorated function
+                if hasattr(target, "__temporal_activity_definition"):
+                    self.logger.debug(
+                        f"target {name} has __temporal_activity_definition"
+                    )
+                    return target  # Already decorated with @activity
+
                 if isinstance(target, MethodType):
                     self_ref = target.__self__
 

src/mcp_agent/cli/cloud/main.py

@@ -13,7 +13,13 @@
 from rich.panel import Panel
 from typer.core import TyperGroup
 
-from mcp_agent.cli.cloud.commands import configure_app, deploy_config, login, logout, whoami
+from mcp_agent.cli.cloud.commands import (
+    configure_app,
+    deploy_config,
+    login,
+    logout,
+    whoami,
+)
 from mcp_agent.cli.cloud.commands.app import (
     delete_app,
     get_app_status,

src/mcp_agent/core/context.py

@@ -89,6 +89,10 @@ class Context(BaseModel):
     # Token counting and cost tracking
     token_counter: Optional[TokenCounter] = None
 
+    # Dynamic gateway configuration (per-run overrides via Temporal memo)
+    gateway_url: str | None = None
+    gateway_token: str | None = None
+
     model_config = ConfigDict(
         extra="allow",
         arbitrary_types_allowed=True,  # Tell Pydantic to defer type evaluation

src/mcp_agent/executor/temporal/__init__.py

@@ -32,10 +32,13 @@
 
 from mcp_agent.config import TemporalSettings
 from mcp_agent.executor.executor import Executor, ExecutorConfig, R
+
 from mcp_agent.executor.temporal.workflow_signal import TemporalSignalHandler
 from mcp_agent.executor.workflow_signal import SignalHandler
 from mcp_agent.logging.logger import get_logger
 from mcp_agent.utils.common import unwrap
+from mcp_agent.executor.temporal.system_activities import SystemActivities
+from mcp_agent.executor.temporal.interceptor import ContextPropagationInterceptor
 
 if TYPE_CHECKING:
     from mcp_agent.app import MCPApp
@@ -263,9 +266,9 @@ async def ensure_client(self):
                 api_key=self.config.api_key,
                 tls=self.config.tls,
                 data_converter=pydantic_data_converter,
-                interceptors=[TracingInterceptor()]
+                interceptors=[TracingInterceptor(), ContextPropagationInterceptor()]
                 if self.context.tracing_enabled
-                else [],
+                else [ContextPropagationInterceptor()],
                 rpc_metadata=self.config.rpc_metadata or {},
             )
 
@@ -278,6 +281,7 @@ async def start_workflow(
         wait_for_result: bool = False,
         workflow_id: str | None = None,
         task_queue: str | None = None,
+        workflow_memo: Dict[str, Any] | None = None,
         **kwargs: Any,
     ) -> WorkflowHandle:
         """
@@ -369,6 +373,7 @@ async def start_workflow(
                 task_queue=task_queue,
                 id_reuse_policy=id_reuse_policy,
                 rpc_metadata=self.config.rpc_metadata or {},
+                memo=workflow_memo or {},
             )
         else:
             handle: WorkflowHandle = await self.client.start_workflow(
@@ -377,6 +382,7 @@ async def start_workflow(
                 task_queue=task_queue,
                 id_reuse_policy=id_reuse_policy,
                 rpc_metadata=self.config.rpc_metadata or {},
+                memo=workflow_memo or {},
             )
 
         # Wait for the result if requested
@@ -497,6 +503,13 @@ async def create_temporal_worker_for_app(app: "MCPApp"):
         # Collect activities from the global registry
         activity_registry = running_app.context.task_registry
 
+        # Register system activities (logging, human input proxy, generic relays)
+        sys_acts = SystemActivities(context=running_app.context)
+        app.workflow_task(name="mcp_forward_log")(sys_acts.forward_log)
+        app.workflow_task(name="mcp_request_user_input")(sys_acts.request_user_input)
+        app.workflow_task(name="mcp_relay_notify")(sys_acts.relay_notify)
+        app.workflow_task(name="mcp_relay_request")(sys_acts.relay_request)
+
         for name in activity_registry.list_activities():
             activities.append(activity_registry.get_activity(name))