From d1be7f25319ffae55d92d47a6c6042810b3403b1 Mon Sep 17 00:00:00 2001 From: deci Date: Wed, 10 Dec 2025 16:05:53 -0600 Subject: [PATCH 1/5] #84 - Projects support in MCP, A2A, API --- docs/connectivity.md | 93 ++++++++++++ python/api/api_message.py | 15 +- python/helpers/fasta2a_server.py | 30 +++- python/helpers/mcp_server.py | 139 ++++++++++++++++-- .../settings/a2a/a2a-connection.html | 53 ++++++- .../settings/external/api-examples.html | 62 +++++++- .../settings/mcp/server/example.html | 83 ++++++++--- 7 files changed, 435 insertions(+), 40 deletions(-) diff --git a/docs/connectivity.md b/docs/connectivity.md index 8cfbe250e..a6f465eab 100644 --- a/docs/connectivity.md +++ b/docs/connectivity.md @@ -25,6 +25,7 @@ Send messages to Agent Zero and receive responses. Supports text messages, file * `message` (string, required): The message to send * `attachments` (array, optional): Array of `{filename, base64}` objects * `lifetime_hours` (number, optional): Chat lifetime in hours (default: 24) +* `project` (string, optional): Project name to activate (only on first message) **Headers:** * `X-API-KEY` (required) @@ -169,6 +170,63 @@ async function sendWithAttachment() { sendWithAttachment(); ``` +#### Project Usage Example + +```javascript +// Working with projects +async function sendMessageWithProject() { + try { + // First message - activate project + const response = await fetch('YOUR_AGENT_ZERO_URL/api_message', { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + 'X-API-KEY': 'YOUR_API_KEY' + }, + body: JSON.stringify({ + message: "Analyze the project structure", + project: "my-web-app" // Activates this project + }) + }); + + const data = await response.json(); + + if (response.ok) { + console.log('✅ Project activated!'); + console.log('Context ID:', data.context_id); + console.log('Response:', data.response); + + // Continue conversation - project already set + const followUp = await fetch('YOUR_AGENT_ZERO_URL/api_message', { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + 'X-API-KEY': 'YOUR_API_KEY' + }, + body: JSON.stringify({ + context_id: data.context_id, + message: "What files are in the project?" + // Do NOT include project field here - already set on first message + }) + }); + + const followUpData = await followUp.json(); + console.log('Follow-up response:', followUpData.response); + return followUpData; + } else { + console.error('❌ Error:', data.error); + return null; + } + } catch (error) { + console.error('❌ Request failed:', error); + return null; + } +} + +// Call the function +sendMessageWithProject(); +``` + --- ## `GET/POST /api_log_get` @@ -568,6 +626,30 @@ Below is an example of a `mcp.json` configuration file that a client could use t } ``` +### Project Support in MCP + +You can specify a project for MCP connections by including it in the URL path: + +```json +{ + "mcpServers": { + "agent-zero-with-project": { + "type": "sse", + "url": "YOUR_AGENT_ZERO_URL/mcp/t-YOUR_API_TOKEN/p-my-project-name/sse" + }, + "agent-zero-http-with-project": { + "type": "streamable-http", + "url": "YOUR_AGENT_ZERO_URL/mcp/t-YOUR_API_TOKEN/p-my-project-name/http/" + } + } +} +``` + +When a project is specified in the URL: +- All new chats will be created within that project context +- The agent will have access to project-specific instructions, knowledge, and file structure +- Attempting to use an existing chat_id from a different project will result in an error + --- ## A2A (Agent-to-Agent) Connectivity @@ -583,3 +665,14 @@ To connect another agent to your Agent Zero instance, use the following URL form ``` YOUR_AGENT_ZERO_URL/a2a/t-YOUR_API_TOKEN ``` + +To connect with a specific project active: + +``` +YOUR_AGENT_ZERO_URL/a2a/t-YOUR_API_TOKEN/p-PROJECT_NAME +``` + +When a project is specified: +- All A2A conversations will run in the context of that project +- The agent will have access to project-specific resources, instructions, and knowledge +- This enables project-isolated agent-to-agent communication diff --git a/python/api/api_message.py b/python/api/api_message.py index 385d556dd..f955024d2 100644 --- a/python/api/api_message.py +++ b/python/api/api_message.py @@ -3,7 +3,7 @@ import os from datetime import datetime, timedelta from agent import AgentContext, UserMessage, AgentContextType from python.helpers.api import ApiHandler, Request, Response -from python.helpers import files +from python.helpers import files, projects from python.helpers.print_style import PrintStyle from werkzeug.utils import secure_filename from initialize import initialize_agent @@ -33,6 +33,7 @@ class ApiMessage(ApiHandler): message = input.get("message", "") attachments = input.get("attachments", []) lifetime_hours = input.get("lifetime_hours", 24) # Default 24 hours + project = input.get("project", None) # Optional project name if not message: return Response('{"error": "Message is required"}', status=400, mimetype="application/json") @@ -71,12 +72,24 @@ class ApiMessage(ApiHandler): context = AgentContext.use(context_id) if not context: return Response('{"error": "Context not found"}', status=404, mimetype="application/json") + + # Validation: if project is provided but context already has different project + existing_project = context.get_data(projects.CONTEXT_DATA_KEY_PROJECT) + if project and existing_project and existing_project != project: + return Response('{"error": "Project can only be set on first message"}', status=400, mimetype="application/json") else: config = initialize_agent() context = AgentContext(config=config, type=AgentContextType.USER) AgentContext.use(context.id) context_id = context.id + # Activate project if provided + if project: + try: + projects.activate_project(context_id, project) + except Exception as e: + return Response(f'{{"error": "Failed to activate project: {str(e)}"}}', status=400, mimetype="application/json") + # Update chat lifetime with self._cleanup_lock: self._chat_lifetimes[context_id] = datetime.now() + timedelta(hours=lifetime_hours) diff --git a/python/helpers/fasta2a_server.py b/python/helpers/fasta2a_server.py index d058d1b68..50d4a4908 100644 --- a/python/helpers/fasta2a_server.py +++ b/python/helpers/fasta2a_server.py @@ -5,8 +5,9 @@ import atexit from typing import Any, List import contextlib import threading +import contextvars -from python.helpers import settings +from python.helpers import settings, projects from starlette.requests import Request # Local imports @@ -60,6 +61,9 @@ except ImportError: # pragma: no cover – library not installed _PRINTER = PrintStyle(italic=True, font_color="purple", padding=False) +# Context variable to store project name from URL +_a2a_project_name: contextvars.ContextVar[str | None] = contextvars.ContextVar('a2a_project_name', default=None) + class AgentZeroWorker(Worker): # type: ignore[misc] """Agent Zero implementation of FastA2A Worker.""" @@ -84,6 +88,16 @@ class AgentZeroWorker(Worker): # type: ignore[misc] cfg = initialize_agent() context = AgentContext(cfg, type=AgentContextType.BACKGROUND) + # Activate project if specified in URL + project_name = _a2a_project_name.get() + if project_name: + try: + projects.activate_project(context.id, project_name) + _PRINTER.print(f"[A2A] Activated project: {project_name}") + except Exception as e: + _PRINTER.print(f"[A2A] Failed to activate project: {e}") + raise Exception(f"Failed to activate project: {str(e)}") + # Log user message so it appears instantly in UI chat window context.log.log( type="user", # type: ignore[arg-type] @@ -424,6 +438,9 @@ class DynamicA2AProxy: if path.startswith('/a2a'): path = path[4:] # Remove '/a2a' prefix + # Initialize project name + project_name = None + # Check if path matches token pattern /t-{token}/ if path.startswith('/t-'): # Extract token from path @@ -431,6 +448,14 @@ class DynamicA2AProxy: path_parts = path[3:].split('/', 1) # Remove '/t-' prefix request_token = path_parts[0] remaining_path = '/' + path_parts[1] if len(path_parts) > 1 else '/' + + # Check for project pattern /p-{project}/ + if remaining_path.startswith('/p-'): + project_parts = remaining_path[3:].split('/', 1) + if project_parts[0]: + project_name = project_parts[0] + remaining_path = '/' + project_parts[1] if len(project_parts) > 1 else '/' + _PRINTER.print(f"[A2A] Extracted project from URL: {project_name}") else: request_token = path[3:] remaining_path = '/' @@ -452,6 +477,9 @@ class DynamicA2AProxy: }) return + # Set project name in context variable for use by worker + _a2a_project_name.set(project_name) + # Update scope with cleaned path scope = dict(scope) scope['path'] = remaining_path diff --git a/python/helpers/mcp_server.py b/python/helpers/mcp_server.py index 4c080da69..466cc2277 100644 --- a/python/helpers/mcp_server.py +++ b/python/helpers/mcp_server.py @@ -3,23 +3,31 @@ from typing import Annotated, Literal, Union from urllib.parse import urlparse from openai import BaseModel from pydantic import Field -from fastmcp import FastMCP +from fastmcp import FastMCP # type: ignore +import contextvars from agent import AgentContext, AgentContextType, UserMessage from python.helpers.persist_chat import remove_chat from initialize import initialize_agent from python.helpers.print_style import PrintStyle -from python.helpers import settings +from python.helpers import settings, projects from starlette.middleware import Middleware from starlette.middleware.base import BaseHTTPMiddleware from starlette.exceptions import HTTPException as StarletteHTTPException from starlette.types import ASGIApp, Receive, Scope, Send -from fastmcp.server.http import create_sse_app +from fastmcp.server.http import create_sse_app # type: ignore from starlette.requests import Request import threading _PRINTER = PrintStyle(italic=True, font_color="green", padding=False) +# Context variable to store project name from URL (per-request) +_mcp_project_name: contextvars.ContextVar[str | None] = contextvars.ContextVar('mcp_project_name', default=None) + +# Session storage for project names (persists across SSE tool calls) +# Key: connection identifier, Value: project name +_mcp_project_sessions: dict[str, str | None] = {} +_mcp_session_lock = threading.Lock() mcp_server: FastMCP = FastMCP( name="Agent Zero integrated MCP Server", @@ -127,6 +135,25 @@ async def send_message( description="The response from the remote Agent Zero Instance", title="response" ), ]: + # Get project name from session storage (persists across SSE connection) + # First try context variable (for HTTP requests), then session storage (for SSE tool calls) + project_name = _mcp_project_name.get() + + # If not in context variable, try session storage using current token + if not project_name: + cfg = settings.get_settings() + current_token = cfg.get("mcp_server_token") + _PRINTER.print(f"[MCP] send_message - Looking for project. Token: '{current_token}' (type: {type(current_token).__name__})") + if current_token: + with _mcp_session_lock: + _PRINTER.print(f"[MCP] Session storage keys: {list(_mcp_project_sessions.keys())}") + _PRINTER.print(f"[MCP] Session storage: {_mcp_project_sessions}") + project_name = _mcp_project_sessions.get(current_token) + if project_name: + _PRINTER.print(f"[MCP] Retrieved project from session: {project_name}") + else: + _PRINTER.print(f"[MCP] No project found in session for token: {current_token}") + context: AgentContext | None = None if chat_id: context = AgentContext.get(chat_id) @@ -137,10 +164,27 @@ async def send_message( # whether we should save the chat or delete it afterwards # If we continue a conversation, it must be persistent persistent_chat = True + + # Validation: if project is in URL but context has different project + if project_name: + existing_project = context.get_data(projects.CONTEXT_DATA_KEY_PROJECT) + if existing_project and existing_project != project_name: + return ToolError( + error=f"Chat belongs to project '{existing_project}' but URL specifies '{project_name}'", + chat_id=chat_id + ) else: config = initialize_agent() context = AgentContext(config=config, type=AgentContextType.BACKGROUND) + # Activate project if specified in URL + if project_name: + try: + projects.activate_project(context.id, project_name) + _PRINTER.print(f"[MCP] Activated project: {project_name}") + except Exception as e: + return ToolError(error=f"Failed to activate project: {str(e)}", chat_id="") + if not message: return ToolError( error="Message is required", chat_id=context.id if persistent_chat else "" @@ -325,10 +369,10 @@ class DynamicMcpProxy: def _create_custom_http_app(self, streamable_http_path, auth_server_provider, auth_settings, debug, routes): """Create a custom HTTP app that manages the session manager manually.""" - from fastmcp.server.http import setup_auth_middleware_and_routes, create_base_app - from mcp.server.streamable_http_manager import StreamableHTTPSessionManager + from fastmcp.server.http import setup_auth_middleware_and_routes, create_base_app # type: ignore + from mcp.server.streamable_http_manager import StreamableHTTPSessionManager # type: ignore from starlette.routing import Mount - from mcp.server.auth.middleware.bearer_auth import RequireAuthMiddleware + from mcp.server.auth.middleware.bearer_auth import RequireAuthMiddleware # type: ignore import anyio server_routes = [] @@ -408,12 +452,47 @@ class DynamicMcpProxy: # Route based on path path = scope.get("path", "") - if f"/t-{self.token}/sse" in path or f"t-{self.token}/messages" in path: - # Route to SSE app - await sse_app(scope, receive, send) - elif f"/t-{self.token}/http" in path: - # Route to HTTP app - await http_app(scope, receive, send) + # Check for token in path (with or without project segment) + # Patterns: /t-{token}/sse, /t-{token}/p-{project}/sse, etc. + has_token = f"/t-{self.token}/" in path or f"t-{self.token}/" in path + + # Extract project from path BEFORE cleaning (for session storage) + project_name = None + if "/p-" in path: + try: + parts = path.split("/p-") + if len(parts) > 1: + project_part = parts[1].split("/")[0] + if project_part: + project_name = project_part + _PRINTER.print(f"[MCP] Proxy extracted project from URL: {project_name}") + except Exception as e: + _PRINTER.print(f"[MCP] Failed to extract project in proxy: {e}") + + # Store project in session (persists across SSE connection) + if self.token and project_name: + with _mcp_session_lock: + _mcp_project_sessions[self.token] = project_name + _PRINTER.print(f"[MCP] Stored project '{project_name}' for token '{self.token}' (type: {type(self.token).__name__}) in proxy") + + # Strip project segment from path if present (e.g., /p-project_name/) + # This is needed because the underlying MCP apps were configured without project paths + cleaned_path = path + if "/p-" in path: + # Remove /p-{project}/ segment: /t-TOKEN/p-PROJECT/sse -> /t-TOKEN/sse + import re + cleaned_path = re.sub(r'/p-[^/]+/', '/', path) + + # Update scope with cleaned path for the underlying app + modified_scope = dict(scope) + modified_scope['path'] = cleaned_path + + if has_token and ("/sse" in path or "/messages" in path): + # Route to SSE app with cleaned path + await sse_app(modified_scope, receive, send) + elif has_token and "/http" in path: + # Route to HTTP app with cleaned path + await http_app(modified_scope, receive, send) else: raise StarletteHTTPException( status_code=403, detail="MCP forbidden" @@ -430,4 +509,40 @@ async def mcp_middleware(request: Request, call_next): status_code=403, detail="MCP server is disabled in settings." ) + # Extract project from URL path if present (pattern: /mcp/t-{token}/p-{project}/...) + path = request.url.path + project_name = None + token = None + + # Extract token from path + if "/t-" in path: + token_parts = path.split("/t-") + if len(token_parts) > 1: + token = token_parts[1].split("/")[0] + + # Extract project if present + if "/p-" in path: + try: + parts = path.split("/p-") + if len(parts) > 1: + project_part = parts[1].split("/")[0] + if project_part: + project_name = project_part + _PRINTER.print(f"[MCP] Extracted project from URL: {project_name}") + except Exception as e: + _PRINTER.print(f"[MCP] Failed to extract project from URL: {e}") + + # Debug logging + _PRINTER.print(f"[MCP] Middleware - Path: {path}, Token: {token}, Project: {project_name}") + + # Store project in session dict ONLY if we found one (don't overwrite with None) + # The proxy already handles project extraction before path cleaning + if token and project_name: + with _mcp_session_lock: + _mcp_project_sessions[token] = project_name + _PRINTER.print(f"[MCP] Middleware stored project '{project_name}' for token session") + + # Also set in context variable for backwards compatibility + _mcp_project_name.set(project_name) + return await call_next(request) diff --git a/webui/components/settings/a2a/a2a-connection.html b/webui/components/settings/a2a/a2a-connection.html index c71c575f2..fc640a68d 100644 --- a/webui/components/settings/a2a/a2a-connection.html +++ b/webui/components/settings/a2a/a2a-connection.html @@ -21,10 +21,21 @@

A2A Connection URL

+ + +
+ + +
+
diff --git a/webui/components/settings/external/api-examples.html b/webui/components/settings/external/api-examples.html index 08569a1e9..615594682 100644 --- a/webui/components/settings/external/api-examples.html +++ b/webui/components/settings/external/api-examples.html @@ -34,7 +34,8 @@ • context_id (string, optional): Existing chat context ID
• message (string, required): The message to send
• attachments (array, optional): Array of {filename, base64} objects
- • lifetime_hours (number, optional): Chat lifetime in hours (default: 24) + • lifetime_hours (number, optional): Chat lifetime in hours (default: 24)
+ • project (string, optional): Project name to activate (only on first message)

Headers: X-API-KEY (required), Content-Type: application/json @@ -52,6 +53,9 @@

File Attachment Example

+ +

Project Usage Example

+
@@ -602,11 +606,66 @@ async function attachmentWorkflow() { // Run the complete workflow attachmentWorkflow();`; + // Project usage example + const projectExample = `// Working with projects +async function sendMessageWithProject() { + try { + // First message - activate project + const response = await fetch('${url}/api_message', { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + 'X-API-KEY': '${token}' + }, + body: JSON.stringify({ + message: "Analyze the project structure", + project: "my-web-app" // Activates this project + }) + }); + + const data = await response.json(); + + if (response.ok) { + console.log('✅ Project activated!'); + console.log('Context ID:', data.context_id); + console.log('Response:', data.response); + + // Continue conversation - project already set + const followUp = await fetch('${url}/api_message', { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + 'X-API-KEY': '${token}' + }, + body: JSON.stringify({ + context_id: data.context_id, + message: "What files are in the project?" + // Do NOT include project field here - already set on first message + }) + }); + + const followUpData = await followUp.json(); + console.log('Follow-up response:', followUpData.response); + return followUpData; + } else { + console.error('❌ Error:', data.error); + return null; + } + } catch (error) { + console.error('❌ Request failed:', error); + return null; + } +} + +// Call the function +sendMessageWithProject();`; + // Initialize ACE editors const editors = [ { id: "api-basic-example", content: basicExample }, { id: "api-continuation-example", content: continuationExample }, { id: "api-attachment-example", content: attachmentExample }, + { id: "api-project-example", content: projectExample }, { id: "api-log-get-example", content: logGetExample }, { id: "api-log-post-example", content: logPostExample }, { id: "api-terminate-example", content: terminateExample }, @@ -635,6 +694,7 @@ attachmentWorkflow();`; #api-basic-example, #api-continuation-example, #api-attachment-example, + #api-project-example, #api-log-get-example, #api-log-post-example, #api-terminate-example, diff --git a/webui/components/settings/mcp/server/example.html b/webui/components/settings/mcp/server/example.html index 4a1e033a6..db50343f4 100644 --- a/webui/components/settings/mcp/server/example.html +++ b/webui/components/settings/mcp/server/example.html @@ -20,37 +20,78 @@

Example MCP Server Configuration JSON

+ + +
+ + +
+
From 3abcf1965119967d64f2c63d1cc8b8926eb5d16a Mon Sep 17 00:00:00 2001 From: deci Date: Thu, 11 Dec 2025 10:07:55 -0600 Subject: [PATCH 2/5] #84 - Fix MCP/A2A project activation --- python/helpers/fasta2a_server.py | 24 ++++++++--- python/helpers/mcp_server.py | 74 ++++---------------------------- 2 files changed, 26 insertions(+), 72 deletions(-) diff --git a/python/helpers/fasta2a_server.py b/python/helpers/fasta2a_server.py index 50d4a4908..693663258 100644 --- a/python/helpers/fasta2a_server.py +++ b/python/helpers/fasta2a_server.py @@ -5,7 +5,7 @@ import atexit from typing import Any, List import contextlib import threading -import contextvars +from collections import deque from python.helpers import settings, projects from starlette.requests import Request @@ -61,8 +61,10 @@ except ImportError: # pragma: no cover – library not installed _PRINTER = PrintStyle(italic=True, font_color="purple", padding=False) -# Context variable to store project name from URL -_a2a_project_name: contextvars.ContextVar[str | None] = contextvars.ContextVar('a2a_project_name', default=None) +# FIFO queue to pass project names from request context to worker context +# Each request appends project (or None), worker pops in same order +_a2a_project_queue: deque[str | None] = deque() +_a2a_project_lock = threading.Lock() class AgentZeroWorker(Worker): # type: ignore[misc] @@ -88,8 +90,14 @@ class AgentZeroWorker(Worker): # type: ignore[misc] cfg = initialize_agent() context = AgentContext(cfg, type=AgentContextType.BACKGROUND) - # Activate project if specified in URL - project_name = _a2a_project_name.get() + # Retrieve project from queue (FIFO matches task processing order) + project_name = None + with _a2a_project_lock: + if _a2a_project_queue: + project_name = _a2a_project_queue.popleft() + _PRINTER.print(f"[A2A] Retrieved project from queue: {project_name}") + + # Activate project if specified if project_name: try: projects.activate_project(context.id, project_name) @@ -477,8 +485,10 @@ class DynamicA2AProxy: }) return - # Set project name in context variable for use by worker - _a2a_project_name.set(project_name) + # Store project in queue for worker to retrieve (maintains FIFO order) + with _a2a_project_lock: + _a2a_project_queue.append(project_name) # None is valid (no project) + _PRINTER.print(f"[A2A] Appended project to queue: {project_name}") # Update scope with cleaned path scope = dict(scope) diff --git a/python/helpers/mcp_server.py b/python/helpers/mcp_server.py index 466cc2277..8db6913ee 100644 --- a/python/helpers/mcp_server.py +++ b/python/helpers/mcp_server.py @@ -24,11 +24,6 @@ _PRINTER = PrintStyle(italic=True, font_color="green", padding=False) # Context variable to store project name from URL (per-request) _mcp_project_name: contextvars.ContextVar[str | None] = contextvars.ContextVar('mcp_project_name', default=None) -# Session storage for project names (persists across SSE tool calls) -# Key: connection identifier, Value: project name -_mcp_project_sessions: dict[str, str | None] = {} -_mcp_session_lock = threading.Lock() - mcp_server: FastMCP = FastMCP( name="Agent Zero integrated MCP Server", instructions=""" @@ -135,24 +130,10 @@ async def send_message( description="The response from the remote Agent Zero Instance", title="response" ), ]: - # Get project name from session storage (persists across SSE connection) - # First try context variable (for HTTP requests), then session storage (for SSE tool calls) + # Get project name from context variable (set in proxy __call__) project_name = _mcp_project_name.get() - - # If not in context variable, try session storage using current token - if not project_name: - cfg = settings.get_settings() - current_token = cfg.get("mcp_server_token") - _PRINTER.print(f"[MCP] send_message - Looking for project. Token: '{current_token}' (type: {type(current_token).__name__})") - if current_token: - with _mcp_session_lock: - _PRINTER.print(f"[MCP] Session storage keys: {list(_mcp_project_sessions.keys())}") - _PRINTER.print(f"[MCP] Session storage: {_mcp_project_sessions}") - project_name = _mcp_project_sessions.get(current_token) - if project_name: - _PRINTER.print(f"[MCP] Retrieved project from session: {project_name}") - else: - _PRINTER.print(f"[MCP] No project found in session for token: {current_token}") + if project_name: + _PRINTER.print(f"[MCP] send_message using project: {project_name}") context: AgentContext | None = None if chat_id: @@ -456,7 +437,7 @@ class DynamicMcpProxy: # Patterns: /t-{token}/sse, /t-{token}/p-{project}/sse, etc. has_token = f"/t-{self.token}/" in path or f"t-{self.token}/" in path - # Extract project from path BEFORE cleaning (for session storage) + # Extract project from path BEFORE cleaning and set in context variable project_name = None if "/p-" in path: try: @@ -469,11 +450,10 @@ class DynamicMcpProxy: except Exception as e: _PRINTER.print(f"[MCP] Failed to extract project in proxy: {e}") - # Store project in session (persists across SSE connection) - if self.token and project_name: - with _mcp_session_lock: - _mcp_project_sessions[self.token] = project_name - _PRINTER.print(f"[MCP] Stored project '{project_name}' for token '{self.token}' (type: {type(self.token).__name__}) in proxy") + # Store project in context variable (will be available in send_message) + _mcp_project_name.set(project_name) + if project_name: + _PRINTER.print(f"[MCP] Set project in context variable: {project_name}") # Strip project segment from path if present (e.g., /p-project_name/) # This is needed because the underlying MCP apps were configured without project paths @@ -500,7 +480,7 @@ class DynamicMcpProxy: async def mcp_middleware(request: Request, call_next): - + """Middleware to check if MCP server is enabled.""" # check if MCP server is enabled cfg = settings.get_settings() if not cfg["mcp_server_enabled"]: @@ -509,40 +489,4 @@ async def mcp_middleware(request: Request, call_next): status_code=403, detail="MCP server is disabled in settings." ) - # Extract project from URL path if present (pattern: /mcp/t-{token}/p-{project}/...) - path = request.url.path - project_name = None - token = None - - # Extract token from path - if "/t-" in path: - token_parts = path.split("/t-") - if len(token_parts) > 1: - token = token_parts[1].split("/")[0] - - # Extract project if present - if "/p-" in path: - try: - parts = path.split("/p-") - if len(parts) > 1: - project_part = parts[1].split("/")[0] - if project_part: - project_name = project_part - _PRINTER.print(f"[MCP] Extracted project from URL: {project_name}") - except Exception as e: - _PRINTER.print(f"[MCP] Failed to extract project from URL: {e}") - - # Debug logging - _PRINTER.print(f"[MCP] Middleware - Path: {path}, Token: {token}, Project: {project_name}") - - # Store project in session dict ONLY if we found one (don't overwrite with None) - # The proxy already handles project extraction before path cleaning - if token and project_name: - with _mcp_session_lock: - _mcp_project_sessions[token] = project_name - _PRINTER.print(f"[MCP] Middleware stored project '{project_name}' for token session") - - # Also set in context variable for backwards compatibility - _mcp_project_name.set(project_name) - return await call_next(request) From 2c87e17f18ace3c67468ea793fef0b63895a891b Mon Sep 17 00:00:00 2001 From: deci Date: Fri, 12 Dec 2025 11:19:58 -0600 Subject: [PATCH 3/5] #84 - APA passing the project via message_id-based dictionary instead of FIFO queue --- python/helpers/fasta2a_server.py | 73 ++++++++++++++++++++++++++------ 1 file changed, 60 insertions(+), 13 deletions(-) diff --git a/python/helpers/fasta2a_server.py b/python/helpers/fasta2a_server.py index 693663258..613f8fb32 100644 --- a/python/helpers/fasta2a_server.py +++ b/python/helpers/fasta2a_server.py @@ -5,7 +5,6 @@ import atexit from typing import Any, List import contextlib import threading -from collections import deque from python.helpers import settings, projects from starlette.requests import Request @@ -61,9 +60,9 @@ except ImportError: # pragma: no cover – library not installed _PRINTER = PrintStyle(italic=True, font_color="purple", padding=False) -# FIFO queue to pass project names from request context to worker context -# Each request appends project (or None), worker pops in same order -_a2a_project_queue: deque[str | None] = deque() +# Map message_id to project name (thread-safe, no race conditions) +# Each message has unique ID that correlates request to worker +_a2a_message_projects: dict[str, str] = {} _a2a_project_lock = threading.Lock() @@ -90,12 +89,15 @@ class AgentZeroWorker(Worker): # type: ignore[misc] cfg = initialize_agent() context = AgentContext(cfg, type=AgentContextType.BACKGROUND) - # Retrieve project from queue (FIFO matches task processing order) + # Retrieve project by message_id (direct lookup, no race conditions) + # Note: Request has messageId (camelCase), but FastA2A converts it to message_id (snake_case) project_name = None - with _a2a_project_lock: - if _a2a_project_queue: - project_name = _a2a_project_queue.popleft() - _PRINTER.print(f"[A2A] Retrieved project from queue: {project_name}") + message_id = message.get('message_id') # FastA2A converts camelCase to snake_case + if message_id: + with _a2a_project_lock: + project_name = _a2a_message_projects.pop(message_id, None) + if project_name: + _PRINTER.print(f"[A2A] Retrieved project for message {message_id}: {project_name}") # Activate project if specified if project_name: @@ -485,10 +487,55 @@ class DynamicA2AProxy: }) return - # Store project in queue for worker to retrieve (maintains FIFO order) - with _a2a_project_lock: - _a2a_project_queue.append(project_name) # None is valid (no project) - _PRINTER.print(f"[A2A] Appended project to queue: {project_name}") + # If project specified, we need to extract message_id from request body + # to correlate project with the specific message (no race conditions) + if project_name: + # Buffer all messages for replay + received_messages = [] + replay_index = 0 + message_id_extracted = False + original_receive = receive + + async def receive_wrapper(): + nonlocal replay_index, message_id_extracted + + # If replaying buffered messages, return them in order + if replay_index < len(received_messages): + msg = received_messages[replay_index] + replay_index += 1 + return msg + + # Otherwise, receive and buffer the next message + message = await original_receive() + received_messages.append(message) + + # Parse message_id when we get the complete body + if message['type'] == 'http.request': + # If this is the last chunk, parse the full body + if not message.get('more_body', False) and not message_id_extracted: + message_id_extracted = True + try: + import json + # Reconstruct full body from all buffered messages + body_parts = [msg.get('body', b'') for msg in received_messages if msg['type'] == 'http.request'] + full_body = b''.join(body_parts) + data = json.loads(full_body) + # Handle JSON-RPC format: params.message.messageId (camelCase!) + if 'params' in data and 'message' in data['params']: + msg = data['params']['message'] + message_id = msg.get('messageId') # camelCase in raw JSON + if message_id: + with _a2a_project_lock: + _a2a_message_projects[message_id] = project_name + _PRINTER.print(f"[A2A] Stored project '{project_name}' for message {message_id}") + # Reset replay index so FastA2A can replay from start + replay_index = 0 + except Exception as e: + _PRINTER.print(f"[A2A] Failed to parse message_id: {e}") + + return message + + receive = receive_wrapper # Update scope with cleaned path scope = dict(scope) From 31daa48a2a54ddc3383b8822188402e9da723165 Mon Sep 17 00:00:00 2001 From: deci Date: Fri, 12 Dec 2025 12:22:41 -0600 Subject: [PATCH 4/5] #84 - adjusted A2A project passing via message.metadata --- python/helpers/fasta2a_server.py | 90 ++++++++++++++------------------ 1 file changed, 40 insertions(+), 50 deletions(-) diff --git a/python/helpers/fasta2a_server.py b/python/helpers/fasta2a_server.py index 613f8fb32..5ee1e2154 100644 --- a/python/helpers/fasta2a_server.py +++ b/python/helpers/fasta2a_server.py @@ -60,11 +60,6 @@ except ImportError: # pragma: no cover – library not installed _PRINTER = PrintStyle(italic=True, font_color="purple", padding=False) -# Map message_id to project name (thread-safe, no race conditions) -# Each message has unique ID that correlates request to worker -_a2a_message_projects: dict[str, str] = {} -_a2a_project_lock = threading.Lock() - class AgentZeroWorker(Worker): # type: ignore[misc] """Agent Zero implementation of FastA2A Worker.""" @@ -89,15 +84,11 @@ class AgentZeroWorker(Worker): # type: ignore[misc] cfg = initialize_agent() context = AgentContext(cfg, type=AgentContextType.BACKGROUND) - # Retrieve project by message_id (direct lookup, no race conditions) - # Note: Request has messageId (camelCase), but FastA2A converts it to message_id (snake_case) - project_name = None - message_id = message.get('message_id') # FastA2A converts camelCase to snake_case - if message_id: - with _a2a_project_lock: - project_name = _a2a_message_projects.pop(message_id, None) - if project_name: - _PRINTER.print(f"[A2A] Retrieved project for message {message_id}: {project_name}") + # Retrieve project from message.metadata (standard A2A pattern) + metadata = message.get('metadata', {}) or {} + project_name = metadata.get('project') + if project_name: + _PRINTER.print(f"[A2A] Retrieved project from message.metadata: {project_name}") # Activate project if specified if project_name: @@ -487,51 +478,50 @@ class DynamicA2AProxy: }) return - # If project specified, we need to extract message_id from request body - # to correlate project with the specific message (no race conditions) + # If project specified, inject it into the request payload if project_name: - # Buffer all messages for replay + # Buffer messages and modify before returning the complete body received_messages = [] - replay_index = 0 - message_id_extracted = False + body_modified = False original_receive = receive async def receive_wrapper(): - nonlocal replay_index, message_id_extracted + nonlocal body_modified - # If replaying buffered messages, return them in order - if replay_index < len(received_messages): - msg = received_messages[replay_index] - replay_index += 1 - return msg - - # Otherwise, receive and buffer the next message + # Receive and buffer the next message message = await original_receive() received_messages.append(message) - # Parse message_id when we get the complete body - if message['type'] == 'http.request': - # If this is the last chunk, parse the full body - if not message.get('more_body', False) and not message_id_extracted: - message_id_extracted = True - try: - import json - # Reconstruct full body from all buffered messages - body_parts = [msg.get('body', b'') for msg in received_messages if msg['type'] == 'http.request'] - full_body = b''.join(body_parts) - data = json.loads(full_body) - # Handle JSON-RPC format: params.message.messageId (camelCase!) - if 'params' in data and 'message' in data['params']: - msg = data['params']['message'] - message_id = msg.get('messageId') # camelCase in raw JSON - if message_id: - with _a2a_project_lock: - _a2a_message_projects[message_id] = project_name - _PRINTER.print(f"[A2A] Stored project '{project_name}' for message {message_id}") - # Reset replay index so FastA2A can replay from start - replay_index = 0 - except Exception as e: - _PRINTER.print(f"[A2A] Failed to parse message_id: {e}") + # When we get the complete body, inject project into JSON + if message['type'] == 'http.request' and not message.get('more_body', False) and not body_modified: + body_modified = True + try: + import json + # Reconstruct full body from all buffered messages + body_parts = [msg.get('body', b'') for msg in received_messages if msg['type'] == 'http.request'] + full_body = b''.join(body_parts) + data = json.loads(full_body) + + # INJECT project into message.metadata (standard A2A pattern) + if 'params' in data and 'message' in data['params']: + msg_data = data['params']['message'] + # Initialize metadata if it doesn't exist + if 'metadata' not in msg_data or msg_data['metadata'] is None: + msg_data['metadata'] = {} + msg_data['metadata']['project'] = project_name + _PRINTER.print(f"[A2A] Injected project '{project_name}' into message.metadata") + + # Serialize back to JSON + modified_body = json.dumps(data).encode('utf-8') + + # Return modified message IMMEDIATELY (before FastA2A processes it) + return { + 'type': 'http.request', + 'body': modified_body, + 'more_body': False + } + except Exception as e: + _PRINTER.print(f"[A2A] Failed to inject project into payload: {e}") return message From 9c905cdbbdac54af566c16fa3251edb75066bd06 Mon Sep 17 00:00:00 2001 From: frdel <38891707+frdel@users.noreply.github.com> Date: Mon, 15 Dec 2025 14:44:32 +0100 Subject: [PATCH 5/5] cleanup prints in A2A/MCP --- python/helpers/fasta2a_server.py | 12 ++---------- python/helpers/mcp_server.py | 5 ----- 2 files changed, 2 insertions(+), 15 deletions(-) diff --git a/python/helpers/fasta2a_server.py b/python/helpers/fasta2a_server.py index 5ee1e2154..000c5a39d 100644 --- a/python/helpers/fasta2a_server.py +++ b/python/helpers/fasta2a_server.py @@ -87,17 +87,10 @@ class AgentZeroWorker(Worker): # type: ignore[misc] # Retrieve project from message.metadata (standard A2A pattern) metadata = message.get('metadata', {}) or {} project_name = metadata.get('project') - if project_name: - _PRINTER.print(f"[A2A] Retrieved project from message.metadata: {project_name}") - + # Activate project if specified if project_name: - try: - projects.activate_project(context.id, project_name) - _PRINTER.print(f"[A2A] Activated project: {project_name}") - except Exception as e: - _PRINTER.print(f"[A2A] Failed to activate project: {e}") - raise Exception(f"Failed to activate project: {str(e)}") + projects.activate_project(context.id, project_name) # Log user message so it appears instantly in UI chat window context.log.log( @@ -509,7 +502,6 @@ class DynamicA2AProxy: if 'metadata' not in msg_data or msg_data['metadata'] is None: msg_data['metadata'] = {} msg_data['metadata']['project'] = project_name - _PRINTER.print(f"[A2A] Injected project '{project_name}' into message.metadata") # Serialize back to JSON modified_body = json.dumps(data).encode('utf-8') diff --git a/python/helpers/mcp_server.py b/python/helpers/mcp_server.py index 8db6913ee..3c0308ed9 100644 --- a/python/helpers/mcp_server.py +++ b/python/helpers/mcp_server.py @@ -132,8 +132,6 @@ async def send_message( ]: # Get project name from context variable (set in proxy __call__) project_name = _mcp_project_name.get() - if project_name: - _PRINTER.print(f"[MCP] send_message using project: {project_name}") context: AgentContext | None = None if chat_id: @@ -162,7 +160,6 @@ async def send_message( if project_name: try: projects.activate_project(context.id, project_name) - _PRINTER.print(f"[MCP] Activated project: {project_name}") except Exception as e: return ToolError(error=f"Failed to activate project: {str(e)}", chat_id="") @@ -452,8 +449,6 @@ class DynamicMcpProxy: # Store project in context variable (will be available in send_message) _mcp_project_name.set(project_name) - if project_name: - _PRINTER.print(f"[MCP] Set project in context variable: {project_name}") # Strip project segment from path if present (e.g., /p-project_name/) # This is needed because the underlying MCP apps were configured without project paths