#84 - Fix MCP/A2A project activation
deci committed
Dec 11, 2025 at 10:07 UTC
3abcf1965119967d64f2c63d1cc8b8926eb5d16a
2 files changed
+26
-72
python/helpers/fasta2a_server.py
+17
-7
@@ -5,7 +5,7 @@ import atexit
5
from typing import Any, List
6
import contextlib
7
import threading
8
-import contextvars
8
+from collections import deque
9
10
from python.helpers import settings, projects
11
from starlette.requests import Request
@@ -61,8 +61,10 @@ except ImportError: # pragma: no cover – library not installed
61
62
_PRINTER = PrintStyle(italic=True, font_color="purple", padding=False)
63
64
-# Context variable to store project name from URL
65
-_a2a_project_name: contextvars.ContextVar[str | None] = contextvars.ContextVar('a2a_project_name', default=None)
64
+# FIFO queue to pass project names from request context to worker context
65
+# Each request appends project (or None), worker pops in same order
66
+_a2a_project_queue: deque[str | None] = deque()
67
+_a2a_project_lock = threading.Lock()
68
69
70
class AgentZeroWorker(Worker): # type: ignore[misc]
@@ -88,8 +90,14 @@ class AgentZeroWorker(Worker): # type: ignore[misc]
90
cfg = initialize_agent()
91
context = AgentContext(cfg, type=AgentContextType.BACKGROUND)
92
91
- # Activate project if specified in URL
92
- project_name = _a2a_project_name.get()
93
+ # Retrieve project from queue (FIFO matches task processing order)
94
+ project_name = None
95
+ with _a2a_project_lock:
96
+ if _a2a_project_queue:
97
+ project_name = _a2a_project_queue.popleft()
98
+ _PRINTER.print(f"[A2A] Retrieved project from queue: {project_name}")
99
+
100
+ # Activate project if specified
101
if project_name:
102
try:
103
projects.activate_project(context.id, project_name)
@@ -477,8 +485,10 @@ class DynamicA2AProxy:
485
})
486
return
487
480
- # Set project name in context variable for use by worker
481
- _a2a_project_name.set(project_name)
488
+ # Store project in queue for worker to retrieve (maintains FIFO order)
489
+ with _a2a_project_lock:
490
+ _a2a_project_queue.append(project_name) # None is valid (no project)
491
+ _PRINTER.print(f"[A2A] Appended project to queue: {project_name}")
492
493
# Update scope with cleaned path
494
scope = dict(scope)
python/helpers/mcp_server.py
+9
-65
@@ -24,11 +24,6 @@ _PRINTER = PrintStyle(italic=True, font_color="green", padding=False)
24
# Context variable to store project name from URL (per-request)
25
_mcp_project_name: contextvars.ContextVar[str | None] = contextvars.ContextVar('mcp_project_name', default=None)
26
27
-# Session storage for project names (persists across SSE tool calls)
28
-# Key: connection identifier, Value: project name
29
-_mcp_project_sessions: dict[str, str | None] = {}
30
-_mcp_session_lock = threading.Lock()
31
-
27
mcp_server: FastMCP = FastMCP(
28
name="Agent Zero integrated MCP Server",
29
instructions="""
@@ -135,24 +130,10 @@ async def send_message(
130
description="The response from the remote Agent Zero Instance", title="response"
131
),
132
]:
138
- # Get project name from session storage (persists across SSE connection)
139
- # First try context variable (for HTTP requests), then session storage (for SSE tool calls)
133
+ # Get project name from context variable (set in proxy __call__)
134
project_name = _mcp_project_name.get()
141
-
142
- # If not in context variable, try session storage using current token
143
- if not project_name:
144
- cfg = settings.get_settings()
145
- current_token = cfg.get("mcp_server_token")
146
- _PRINTER.print(f"[MCP] send_message - Looking for project. Token: '{current_token}' (type: {type(current_token).__name__})")
147
- if current_token:
148
- with _mcp_session_lock:
149
- _PRINTER.print(f"[MCP] Session storage keys: {list(_mcp_project_sessions.keys())}")
150
- _PRINTER.print(f"[MCP] Session storage: {_mcp_project_sessions}")
151
- project_name = _mcp_project_sessions.get(current_token)
152
- if project_name:
153
- _PRINTER.print(f"[MCP] Retrieved project from session: {project_name}")
154
- else:
155
- _PRINTER.print(f"[MCP] No project found in session for token: {current_token}")
135
+ if project_name:
136
+ _PRINTER.print(f"[MCP] send_message using project: {project_name}")
137
138
context: AgentContext | None = None
139
if chat_id:
@@ -456,7 +437,7 @@ class DynamicMcpProxy:
437
# Patterns: /t-{token}/sse, /t-{token}/p-{project}/sse, etc.
438
has_token = f"/t-{self.token}/" in path or f"t-{self.token}/" in path
439
459
- # Extract project from path BEFORE cleaning (for session storage)
440
+ # Extract project from path BEFORE cleaning and set in context variable
441
project_name = None
442
if "/p-" in path:
443
try:
@@ -469,11 +450,10 @@ class DynamicMcpProxy:
450
except Exception as e:
451
_PRINTER.print(f"[MCP] Failed to extract project in proxy: {e}")
452
472
- # Store project in session (persists across SSE connection)
473
- if self.token and project_name:
474
- with _mcp_session_lock:
475
- _mcp_project_sessions[self.token] = project_name
476
- _PRINTER.print(f"[MCP] Stored project '{project_name}' for token '{self.token}' (type: {type(self.token).__name__}) in proxy")
453
+ # Store project in context variable (will be available in send_message)
454
+ _mcp_project_name.set(project_name)
455
+ if project_name:
456
+ _PRINTER.print(f"[MCP] Set project in context variable: {project_name}")
457
458
# Strip project segment from path if present (e.g., /p-project_name/)
459
# This is needed because the underlying MCP apps were configured without project paths
@@ -500,7 +480,7 @@ class DynamicMcpProxy:
480
481
482
async def mcp_middleware(request: Request, call_next):
503
-
483
+ """Middleware to check if MCP server is enabled."""
484
# check if MCP server is enabled
485
cfg = settings.get_settings()
486
if not cfg["mcp_server_enabled"]:
@@ -509,40 +489,4 @@ async def mcp_middleware(request: Request, call_next):
489
status_code=403, detail="MCP server is disabled in settings."
490
)
491
512
- # Extract project from URL path if present (pattern: /mcp/t-{token}/p-{project}/...)
513
- path = request.url.path
514
- project_name = None
515
- token = None
516
-
517
- # Extract token from path
518
- if "/t-" in path:
519
- token_parts = path.split("/t-")
520
- if len(token_parts) > 1:
521
- token = token_parts[1].split("/")[0]
522
-
523
- # Extract project if present
524
- if "/p-" in path:
525
- try:
526
- parts = path.split("/p-")
527
- if len(parts) > 1:
528
- project_part = parts[1].split("/")[0]
529
- if project_part:
530
- project_name = project_part
531
- _PRINTER.print(f"[MCP] Extracted project from URL: {project_name}")
532
- except Exception as e:
533
- _PRINTER.print(f"[MCP] Failed to extract project from URL: {e}")
534
-
535
- # Debug logging
536
- _PRINTER.print(f"[MCP] Middleware - Path: {path}, Token: {token}, Project: {project_name}")
537
-
538
- # Store project in session dict ONLY if we found one (don't overwrite with None)
539
- # The proxy already handles project extraction before path cleaning
540
- if token and project_name:
541
- with _mcp_session_lock:
542
- _mcp_project_sessions[token] = project_name
543
- _PRINTER.print(f"[MCP] Middleware stored project '{project_name}' for token session")
544
-
545
- # Also set in context variable for backwards compatibility
546
- _mcp_project_name.set(project_name)
547
-
492
return await call_next(request)