diff --git a/src/backend/integrations/internal_mcp/catalogue.py b/src/backend/integrations/internal_mcp/catalogue.py index a210e0c7..b678b34e 100644 --- a/src/backend/integrations/internal_mcp/catalogue.py +++ b/src/backend/integrations/internal_mcp/catalogue.py @@ -9138,191 +9138,127 @@ async def _async_diagnostics(): def _linkedin_proxy_error_response(exc: Exception) -> dict[str, Any]: + error_code = getattr(exc, "error_code", None) or getattr( + exc, "reason_code", None + ) + details = {"exception_type": type(exc).__name__} + error_details = getattr(exc, "details", None) + if isinstance(error_details, Mapping): + details.update(dict(error_details)) return make_error_response( - "linkedin_proxy_error", + str(error_code or "linkedin_proxy_error"), str(exc), - details={"exception_type": type(exc).__name__}, + details=details, suggestions=[ - "Check LinkedIn MCP command/path configuration", - "Check LinkedIn MCP Python dependencies and data-root access", + "Use an authenticated Von account authorised for this LinkedIn resource", + "Check the owner, project, executable, and database settings", ], ) -def _linkedin_list_exports(**kwargs): - from .linkedin_proxy_mcp import get_linkedin_proxy, LinkedInProxyError - - refresh = bool(kwargs.get("refresh", False)) - - async def _async_list_exports(): - proxy = await get_linkedin_proxy() - return await proxy.list_exports(refresh=refresh) - - try: - return _run_async_compat(_async_list_exports) - except LinkedInProxyError as exc: - return _linkedin_proxy_error_response(exc) - except Exception as exc: - return _linkedin_proxy_error_response(exc) - - -def _linkedin_list_files(**kwargs): - from .linkedin_proxy_mcp import get_linkedin_proxy, LinkedInProxyError - - export_name = kwargs.get("export_name") - if not export_name: - return make_error_response( - "missing_parameter", - "Missing required parameter: export_name", - details={"missing": ["export_name"]}, - suggestions=["Call linkedin_list_exports first to discover export names"], - ) - - async def _async_list_files(): - proxy = await get_linkedin_proxy() - return await proxy.list_files(export_name=str(export_name)) - - try: - return _run_async_compat(_async_list_files) - except LinkedInProxyError as exc: - return _linkedin_proxy_error_response(exc) - except Exception as exc: - return _linkedin_proxy_error_response(exc) - - -def _linkedin_get_profile(**kwargs): - from .linkedin_proxy_mcp import get_linkedin_proxy, LinkedInProxyError +def _linkedin_call(tool_name: str, arguments: dict[str, Any], **kwargs): + from .linkedin_proxy_mcp import ( + LinkedInInvocationAuthorityError, + LinkedInProxyError, + get_linkedin_proxy, + resolve_linkedin_invocation_authority, + ) - export_name = kwargs.get("export_name") - if not export_name: + resource_id = kwargs.get("resource_id") + if not isinstance(resource_id, str) or not resource_id.strip(): return make_error_response( "missing_parameter", - "Missing required parameter: export_name", - details={"missing": ["export_name"]}, - suggestions=["Call linkedin_list_exports first to discover export names"], + "Missing trusted LinkedIn resource binding.", + details={"missing": ["resource_id"]}, + suggestions=["Use an authenticated Von conversation authorised for this resource"], ) - async def _async_get_profile(): - proxy = await get_linkedin_proxy() - return await proxy.get_profile(export_name=str(export_name)) + async def _async_call(): + authority = resolve_linkedin_invocation_authority(resource_id=resource_id) + proxy = await get_linkedin_proxy(resource_id=authority.resource_id) + payload = await proxy.call(tool_name, arguments) + payload.setdefault("success", True) + payload["authority"] = authority.receipt() + return payload try: - return _run_async_compat(_async_get_profile) - except LinkedInProxyError as exc: + return _run_async_compat(_async_call) + except (LinkedInInvocationAuthorityError, LinkedInProxyError) as exc: return _linkedin_proxy_error_response(exc) - except Exception as exc: + except Exception as exc: # noqa: BLE001 - translate adapter failures to tool errors return _linkedin_proxy_error_response(exc) -def _linkedin_get_csv_data(**kwargs): - from .linkedin_proxy_mcp import get_linkedin_proxy, LinkedInProxyError - - export_name = kwargs.get("export_name") - file_name = kwargs.get("file_name") - if not export_name or not file_name: - missing = [] - if not export_name: - missing.append("export_name") - if not file_name: - missing.append("file_name") - return make_error_response( - "missing_parameter", - f"Missing required parameter(s): {', '.join(missing)}", - details={"missing": missing}, - suggestions=[ - "Provide export_name and file_name", - "Call linkedin_list_files to discover available CSV files", - ], - ) - - try: - limit = int(kwargs.get("limit", 10)) - except Exception: - return make_error_response( - "invalid_parameter", - "limit must be an integer", - details={"parameter": "limit"}, - ) - limit = max(1, min(200, limit)) - - async def _async_get_csv_data(): - proxy = await get_linkedin_proxy() - return await proxy.get_csv_data( - export_name=str(export_name), - file_name=str(file_name), - limit=limit, - ) +def _linkedin_index_status(**kwargs): + return _linkedin_call("index_status", {}, **kwargs) - try: - return _run_async_compat(_async_get_csv_data) - except LinkedInProxyError as exc: - return _linkedin_proxy_error_response(exc) - except Exception as exc: - return _linkedin_proxy_error_response(exc) +def _linkedin_list_datasets(**kwargs): + return _linkedin_call("list_datasets", {}, **kwargs) -def _linkedin_get_company_stats(**kwargs): - from .linkedin_proxy_mcp import get_linkedin_proxy, LinkedInProxyError - export_name = kwargs.get("export_name") - if not export_name: - return make_error_response( - "missing_parameter", - "Missing required parameter: export_name", - details={"missing": ["export_name"]}, - suggestions=["Call linkedin_list_exports first to discover export names"], - ) - - try: - top_n = int(kwargs.get("top_n", 10)) - except Exception: - return make_error_response( - "invalid_parameter", - "top_n must be an integer", - details={"parameter": "top_n"}, - ) - top_n = max(1, min(200, top_n)) +def _linkedin_search_export(**kwargs): + return _linkedin_call( + "search_export", + { + key: kwargs[key] + for key in ("query", "datasets", "after", "before", "match", "limit") + if kwargs.get(key) is not None + }, + **kwargs, + ) - async def _async_get_company_stats(): - proxy = await get_linkedin_proxy() - return await proxy.get_company_stats(export_name=str(export_name), top_n=top_n) - try: - return _run_async_compat(_async_get_company_stats) - except LinkedInProxyError as exc: - return _linkedin_proxy_error_response(exc) - except Exception as exc: - return _linkedin_proxy_error_response(exc) +def _linkedin_search_connections(**kwargs): + return _linkedin_call( + "search_connections", + { + key: kwargs[key] + for key in ( + "query", + "name", + "company", + "position", + "connected_after", + "connected_before", + "limit", + ) + if kwargs.get(key) is not None + }, + **kwargs, + ) -def _linkedin_get_messages(**kwargs): - from .linkedin_proxy_mcp import get_linkedin_proxy, LinkedInProxyError +def _linkedin_search_messages(**kwargs): + return _linkedin_call( + "search_messages", + { + key: kwargs[key] + for key in ("query", "participant", "after", "before", "limit") + if kwargs.get(key) is not None + }, + **kwargs, + ) - export_name = kwargs.get("export_name") - if not export_name: - return make_error_response( - "missing_parameter", - "Missing required parameter: export_name", - details={"missing": ["export_name"]}, - suggestions=["Call linkedin_list_exports first to discover export names"], - ) - query = kwargs.get("query") - query_text = str(query) if query is not None else "" +def _linkedin_list_connection_organisations(**kwargs): + return _linkedin_call( + "list_connection_organisations", + { + key: kwargs[key] + for key in ("query", "limit") + if kwargs.get(key) is not None + }, + **kwargs, + ) - async def _async_get_messages(): - proxy = await get_linkedin_proxy() - return await proxy.get_messages( - export_name=str(export_name), - query=query_text, - ) - try: - return _run_async_compat(_async_get_messages) - except LinkedInProxyError as exc: - return _linkedin_proxy_error_response(exc) - except Exception as exc: - return _linkedin_proxy_error_response(exc) +def _linkedin_get_record(**kwargs): + return _linkedin_call( + "get_record", + {"record_id": kwargs.get("record_id")}, + **kwargs, + ) def _resilient_extract_url(**kwargs): @@ -12392,69 +12328,103 @@ def _search_proxy_diagnostics_output_schema() -> Schema: ) -def _linkedin_list_exports_input_schema() -> Schema: +def _linkedin_base_input_schema( + *, + description: str, + required: Mapping[str, Any] | None = None, + optional: Mapping[str, Any] | None = None, + enum_values: Mapping[str, tuple[Any, ...]] | None = None, +) -> Schema: return Schema( - required={}, - optional={ - "refresh": (bool, type(None)), - "namespace": (str, type(None)), - }, - allow_unknown=True, - description="linkedin_list_exports input: optional refresh (bool) and namespace (accepted for orchestrator consistency).", + required={"resource_id": str, **dict(required or {})}, + optional=dict(optional or {}), + allow_unknown=False, + description=description, + enum_values=dict(enum_values or {}), ) -def _linkedin_list_files_input_schema() -> Schema: - return Schema( - required={"export_name": str}, - optional={"namespace": (str, type(None))}, - allow_unknown=True, - description="linkedin_list_files input: export_name (str, required), namespace (optional, ignored).", +def _linkedin_index_status_input_schema() -> Schema: + return _linkedin_base_input_schema( + description="linkedin_index_status input: private resource binding supplied by Von.", ) -def _linkedin_get_profile_input_schema() -> Schema: - return Schema( - required={"export_name": str}, - optional={"namespace": (str, type(None))}, - allow_unknown=True, - description="linkedin_get_profile input: export_name (str, required), namespace (optional, ignored).", +def _linkedin_list_datasets_input_schema() -> Schema: + return _linkedin_base_input_schema( + description="linkedin_list_datasets input: private resource binding supplied by Von.", ) -def _linkedin_get_csv_data_input_schema() -> Schema: - return Schema( - required={"export_name": str, "file_name": str}, +def _linkedin_search_export_input_schema() -> Schema: + return _linkedin_base_input_schema( + required={"query": str}, optional={ + "datasets": (list, type(None)), + "after": (str, type(None)), + "before": (str, type(None)), + "match": (str, type(None)), "limit": (int, type(None)), - "namespace": (str, type(None)), }, - allow_unknown=True, - description="linkedin_get_csv_data input: export_name (str), file_name (str), optional limit (int, default 10), namespace (optional, ignored).", + enum_values={"match": ("all", "any")}, + description=( + "linkedin_search_export input: full-text query with optional datasets, " + "ISO date bounds, all/any matching, and bounded result limit." + ), ) -def _linkedin_get_company_stats_input_schema() -> Schema: - return Schema( - required={"export_name": str}, +def _linkedin_search_connections_input_schema() -> Schema: + return _linkedin_base_input_schema( optional={ - "top_n": (int, type(None)), - "namespace": (str, type(None)), + "query": (str, type(None)), + "name": (str, type(None)), + "company": (str, type(None)), + "position": (str, type(None)), + "connected_after": (str, type(None)), + "connected_before": (str, type(None)), + "limit": (int, type(None)), }, - allow_unknown=True, - description="linkedin_get_company_stats input: export_name (str), optional top_n (int, default 10), namespace (optional, ignored).", + description=( + "linkedin_search_connections input: at least one of query, name, company, " + "or position, with optional connection-date bounds and limit." + ), ) -def _linkedin_get_messages_input_schema() -> Schema: - return Schema( - required={"export_name": str}, +def _linkedin_search_messages_input_schema() -> Schema: + return _linkedin_base_input_schema( optional={ "query": (str, type(None)), - "namespace": (str, type(None)), + "participant": (str, type(None)), + "after": (str, type(None)), + "before": (str, type(None)), + "limit": (int, type(None)), }, - allow_unknown=True, - description="linkedin_get_messages input: export_name (str), optional query (str), namespace (optional, ignored).", + description=( + "linkedin_search_messages input: query or participant with optional ISO " + "date bounds and limit. Returns bounded snippets, not whole conversations." + ), + ) + + +def _linkedin_list_connection_organisations_input_schema() -> Schema: + return _linkedin_base_input_schema( + optional={ + "query": (str, type(None)), + "limit": (int, type(None)), + }, + description=( + "linkedin_list_connection_organisations input: optional organisation-name " + "filter and bounded limit." + ), + ) + + +def _linkedin_get_record_input_schema() -> Schema: + return _linkedin_base_input_schema( + required={"record_id": int}, + description="linkedin_get_record input: record id returned by a prior search.", ) @@ -12467,6 +12437,7 @@ def _linkedin_generic_output_schema( "success": (bool, type(None)), "error": (str, type(None)), "error_code": (str, type(None)), + "authority": (dict, type(None)), } if isinstance(optional_fields, Mapping): optional.update(dict(optional_fields)) @@ -12478,68 +12449,35 @@ def _linkedin_generic_output_schema( ) -def _linkedin_list_exports_output_schema() -> Schema: - return _linkedin_generic_output_schema( - description="linkedin_list_exports output: exports list, data_root metadata, or error details.", - optional_fields={ - "exports": (list, type(None)), - "total_exports": (int, type(None)), - "data_root": (str, type(None)), - "data_root_exists": (bool, type(None)), - }, - ) - - -def _linkedin_list_files_output_schema() -> Schema: - return _linkedin_generic_output_schema( - description="linkedin_list_files output: files list for a chosen export, or error details.", - optional_fields={ - "export_name": (str, type(None)), - "files": (list, type(None)), - }, - ) - - -def _linkedin_get_profile_output_schema() -> Schema: - return _linkedin_generic_output_schema( - description="linkedin_get_profile output: profile text for an export, or error details.", - optional_fields={ - "export_name": (str, type(None)), - "profile": (str, type(None)), - }, - ) - - -def _linkedin_get_csv_data_output_schema() -> Schema: - return _linkedin_generic_output_schema( - description="linkedin_get_csv_data output: sampled CSV text content, or error details.", - optional_fields={ - "export_name": (str, type(None)), - "file_name": (str, type(None)), - "limit": (int, type(None)), - "data": (str, type(None)), - }, - ) - - -def _linkedin_get_company_stats_output_schema() -> Schema: +def _linkedin_index_status_output_schema() -> Schema: return _linkedin_generic_output_schema( - description="linkedin_get_company_stats output: company frequency mapping, or error details.", + description="LinkedIn index readiness, source snapshot, counts, and authority receipt.", optional_fields={ - "export_name": (str, type(None)), - "top_n": (int, type(None)), - "company_stats": (dict, type(None)), + "ready": (bool, type(None)), + "schema_version": (str, type(None)), + "source_archive": (str, type(None)), + "source_archive_sha256": (str, type(None)), + "built_at": (str, type(None)), + "datasets": (int, type(None)), + "records": (int, type(None)), }, ) -def _linkedin_get_messages_output_schema() -> Schema: +def _linkedin_search_output_schema(description: str) -> Schema: return _linkedin_generic_output_schema( - description="linkedin_get_messages output: message sample text (optionally filtered), or error details.", + description=description, optional_fields={ - "export_name": (str, type(None)), "query": (str, type(None)), - "messages": (str, type(None)), + "count": (int, type(None)), + "limit": (int, type(None)), + "results": (list, type(None)), + "filters": (dict, type(None)), + "datasets": (list, type(None)), + "organisations": (list, type(None)), + "record": (dict, type(None)), + "record_id": (int, type(None)), + "source": (dict, type(None)), }, ) @@ -38126,80 +38064,124 @@ def _build_default_catalogue_knowledge_io_definitions() -> List[MethodDefinition "inclusion for direct jira_add_attachment calls." ), ), - # LinkedIn Data Dump MCP tools (local external server) + # Owner-scoped LinkedIn export search via a local external MCP server. MethodDefinition( - name="linkedin_list_exports", - handler=_linkedin_list_exports, - input_schema=_linkedin_list_exports_input_schema(), - output_schema=_linkedin_list_exports_output_schema(), + name="linkedin_index_status", + handler=_linkedin_index_status, + input_schema=_linkedin_index_status_input_schema(), + output_schema=_linkedin_index_status_output_schema(), category="read", - ordinary_turn_excluded_reason="host_local_private_data", - timeout_sec=20.0, + ordinary_turn_trusted_argument_bindings={ + "resource_id": "linkedin_resource_id", + }, + advisory_timeout_sec=10.0, description=( - "List available LinkedIn data exports from the configured local data root. " - "Use this first to discover valid export_name values for subsequent LinkedIn tools." + "Check the authorised private LinkedIn index snapshot and readiness. " + "Returns source provenance and an authority receipt." ), ), MethodDefinition( - name="linkedin_list_files", - handler=_linkedin_list_files, - input_schema=_linkedin_list_files_input_schema(), - output_schema=_linkedin_list_files_output_schema(), + name="linkedin_list_datasets", + handler=_linkedin_list_datasets, + input_schema=_linkedin_list_datasets_input_schema(), + output_schema=_linkedin_search_output_schema( + "Indexed LinkedIn datasets and their source members." + ), category="read", - ordinary_turn_excluded_reason="host_local_private_data", - timeout_sec=20.0, + ordinary_turn_trusted_argument_bindings={ + "resource_id": "linkedin_resource_id", + }, + advisory_timeout_sec=10.0, description=( - "List files within one LinkedIn export. " - "Use after linkedin_list_exports to discover available CSV/file names." + "List datasets in the authenticated actor's LinkedIn export. " + "Use when the relevant export surface is not already known." ), ), MethodDefinition( - name="linkedin_get_profile", - handler=_linkedin_get_profile, - input_schema=_linkedin_get_profile_input_schema(), - output_schema=_linkedin_get_profile_output_schema(), + name="linkedin_search_export", + handler=_linkedin_search_export, + input_schema=_linkedin_search_export_input_schema(), + output_schema=_linkedin_search_output_schema( + "Bounded full-text LinkedIn export search with provenance." + ), category="read", - ordinary_turn_excluded_reason="host_local_private_data", - timeout_sec=20.0, + ordinary_turn_trusted_argument_bindings={ + "resource_id": "linkedin_resource_id", + }, + advisory_timeout_sec=15.0, + description=( + "Search the authenticated actor's private LinkedIn export across indexed " + "datasets. Use narrow terms and inspect a full record only when needed." + ), + ), + MethodDefinition( + name="linkedin_search_connections", + handler=_linkedin_search_connections, + input_schema=_linkedin_search_connections_input_schema(), + output_schema=_linkedin_search_output_schema( + "Structured LinkedIn connection matches with source provenance." + ), + category="read", + ordinary_turn_trusted_argument_bindings={ + "resource_id": "linkedin_resource_id", + }, + advisory_timeout_sec=15.0, description=( - "Get profile information from Profile.csv for a selected LinkedIn export." + "Find the authenticated actor's LinkedIn connections by name, company, " + "position, free text, or connection date. Employer does not prove a " + "person's current location." ), ), MethodDefinition( - name="linkedin_get_csv_data", - handler=_linkedin_get_csv_data, - input_schema=_linkedin_get_csv_data_input_schema(), - output_schema=_linkedin_get_csv_data_output_schema(), + name="linkedin_search_messages", + handler=_linkedin_search_messages, + input_schema=_linkedin_search_messages_input_schema(), + output_schema=_linkedin_search_output_schema( + "Bounded private LinkedIn message snippets with source provenance." + ), category="read", - ordinary_turn_excluded_reason="host_local_private_data", - timeout_sec=25.0, + ordinary_turn_trusted_argument_bindings={ + "resource_id": "linkedin_resource_id", + }, + advisory_timeout_sec=15.0, description=( - "Read sampled rows from any CSV file in a LinkedIn export " - "(for example Education.csv, Languages.csv, Publications.csv)." + "Search the authenticated actor's private LinkedIn messages by text, " + "participant, or date. Returns snippets; retrieve a full record only " + "when the request needs it." ), ), MethodDefinition( - name="linkedin_get_company_stats", - handler=_linkedin_get_company_stats, - input_schema=_linkedin_get_company_stats_input_schema(), - output_schema=_linkedin_get_company_stats_output_schema(), + name="linkedin_list_connection_organisations", + handler=_linkedin_list_connection_organisations, + input_schema=_linkedin_list_connection_organisations_input_schema(), + output_schema=_linkedin_search_output_schema( + "Connection employer counts recorded in the LinkedIn export." + ), category="read", - ordinary_turn_excluded_reason="host_local_private_data", - timeout_sec=25.0, + ordinary_turn_trusted_argument_bindings={ + "resource_id": "linkedin_resource_id", + }, + advisory_timeout_sec=10.0, description=( - "Get top company counts from Connections.csv for a LinkedIn export." + "List employers recorded for the authenticated actor's LinkedIn " + "connections, optionally filtered by organisation name." ), ), MethodDefinition( - name="linkedin_get_messages", - handler=_linkedin_get_messages, - input_schema=_linkedin_get_messages_input_schema(), - output_schema=_linkedin_get_messages_output_schema(), + name="linkedin_get_record", + handler=_linkedin_get_record, + input_schema=_linkedin_get_record_input_schema(), + output_schema=_linkedin_search_output_schema( + "One deliberately selected full LinkedIn source record." + ), category="read", - ordinary_turn_excluded_reason="host_local_private_data", - timeout_sec=25.0, + ordinary_turn_trusted_argument_bindings={ + "resource_id": "linkedin_resource_id", + }, + advisory_timeout_sec=10.0, description=( - "Retrieve message rows from a LinkedIn export (optionally filtered by query text)." + "Retrieve one full private LinkedIn source row by an id returned from a " + "search. Do not use for broad discovery." ), ), # Search MCP tools (Tavily) diff --git a/src/backend/integrations/internal_mcp/linkedin_proxy_mcp.py b/src/backend/integrations/internal_mcp/linkedin_proxy_mcp.py index 34fb23f7..9d6b90aa 100644 --- a/src/backend/integrations/internal_mcp/linkedin_proxy_mcp.py +++ b/src/backend/integrations/internal_mcp/linkedin_proxy_mcp.py @@ -1,101 +1,239 @@ -"""LinkedIn Data Dump MCP proxy using the shared stdio client helper. +"""Owner-scoped proxy for the local LinkedIn export MCP server. -This proxy bridges a local LinkedIn MCP server so Von's internal MCP catalogue -can expose deterministic, typed tools for LinkedIn export inspection. +Von authenticates and authorises the actor. The external stdio process is +deliberately simpler: each process is bound to one read-only SQLite index via +``LINKEDIN_MJW_DB`` and receives neither Google credentials nor a caller- +supplied database path. """ from __future__ import annotations -import ast -import json import logging import os +import shutil import threading -from dataclasses import dataclass +from collections.abc import Mapping +from dataclasses import dataclass, field from pathlib import Path -from typing import Any, Dict, Optional +from typing import Any, Literal +from ...utils.runtime_env import clean_env_value, read_repo_dotenv_values from .mcp_proxy_base import MCPServerConfig, MCPStdIOClient, MCPToolClientError logger = logging.getLogger(__name__) _LOG_TAG = "[linkedin_proxy]" -_DEFAULT_SERVER_PATH = Path("LinkedInMCP") / "server.py" -_DEFAULT_PYTHONPATH = _DEFAULT_SERVER_PATH.parent -_DEFAULT_DATA_ROOT = Path("data") / "linkedin" +LINKEDIN_RESOURCE_ID_ENV = "VON_LINKEDIN_RESOURCE_ID" +LINKEDIN_OWNER_USER_CONCEPT_ID_ENV = "VON_LINKEDIN_OWNER_USER_CONCEPT_ID" +LINKEDIN_PROJECT_DIR_ENV = "VON_LINKEDIN_MCP_PROJECT_DIR" +LINKEDIN_COMMAND_ENV = "VON_LINKEDIN_MCP_COMMAND" +LINKEDIN_TIMEOUT_ENV = "VON_LINKEDIN_MCP_TIMEOUT_SEC" +LINKEDIN_DATABASE_ENV = "LINKEDIN_MJW_DB" + +DEFAULT_RESOURCE_ID = "personal_linkedin" +_DOTENV_KEYS = ( + LINKEDIN_RESOURCE_ID_ENV, + LINKEDIN_OWNER_USER_CONCEPT_ID_ENV, + LINKEDIN_PROJECT_DIR_ENV, + LINKEDIN_COMMAND_ENV, + LINKEDIN_TIMEOUT_ENV, + LINKEDIN_DATABASE_ENV, +) +_SUBPROCESS_PASSTHROUGH_KEYS = ( + "PATH", + "PATHEXT", + "SYSTEMROOT", + "WINDIR", + "COMSPEC", + "LANG", + "LC_ALL", + "LC_CTYPE", + "SSL_CERT_FILE", + "SSL_CERT_DIR", +) class LinkedInProxyError(Exception): - """Raised when LinkedIn MCP proxy operations fail.""" + """Raised when LinkedIn MCP configuration or transport fails.""" + def __init__( + self, + message: str, + *, + error_code: str = "linkedin_proxy_error", + details: Mapping[str, Any] | None = None, + ) -> None: + super().__init__(message) + self.error_code = error_code + self.details = dict(details or {}) + + +@dataclass(frozen=True) +class LinkedInInvocationAuthorityError(PermissionError): + """Fail-closed denial produced before the private MCP is launched.""" + + reason_code: str + safe_message: str + details: Mapping[str, Any] = field(default_factory=dict) + + def __str__(self) -> str: + return self.safe_message -@dataclass -class LinkedInProxyConfig: - """Configuration for LinkedIn MCP subprocess.""" +LinkedInPrincipalKind = Literal["authenticated_actor", "trusted_local_operator"] + + +@dataclass(frozen=True) +class LinkedInInvocationAuthority: + principal_kind: LinkedInPrincipalKind + resource_id: str + actor_user_concept_id: str | None = None + + def receipt(self) -> dict[str, Any]: + return { + "schema_version": "linkedin_resource_authority.v1", + "authorised": True, + "principal_kind": self.principal_kind, + "actor_user_concept_id": self.actor_user_concept_id, + "resource_id": self.resource_id, + "grant_source": ( + "deployment_owner_binding" + if self.principal_kind == "authenticated_actor" + else "trusted_local_operator" + ), + "access": "read_only", + } + + +@dataclass(frozen=True) +class LinkedInProxyConfig: + resource_id: str + owner_user_concept_id: str + project_dir: Path + database: Path command: str args: list[str] - env: Dict[str, str] - data_root: Path + env: dict[str, str] timeout_sec: float = 30.0 -def _coerce_value_from_text_payload(value: Any) -> Any: - """Normalise MCP text payload wrappers into structured values when possible.""" - - if isinstance(value, dict): - text_payload = value.get("text") - if isinstance(text_payload, str): - stripped = text_payload.strip() - if stripped: - try: - return json.loads(stripped) - except Exception: - pass - try: - parsed_literal = ast.literal_eval(stripped) - if isinstance(parsed_literal, (dict, list, str, int, float, bool)): - return parsed_literal - except Exception: - pass - return value - return value - - -def _coerce_list_payload(value: Any) -> list[str]: - payload = _coerce_value_from_text_payload(value) - if isinstance(payload, list): - return [str(item) for item in payload] - if isinstance(payload, dict): - raw_items = payload.get("files") or payload.get("exports") or payload.get("items") - if isinstance(raw_items, list): - return [str(item) for item in raw_items] - return [] - - -def _coerce_mapping_payload(value: Any) -> dict[str, Any]: - payload = _coerce_value_from_text_payload(value) - if isinstance(payload, dict): - return payload - return {} - - -def _coerce_text_payload(value: Any) -> str: - payload = _coerce_value_from_text_payload(value) - if isinstance(payload, str): - return payload - if isinstance(payload, dict): - text = payload.get("text") - if isinstance(text, str): - return text - return json.dumps(payload, ensure_ascii=True, default=str) - if isinstance(payload, list): - return "\n".join(str(item) for item in payload) - return str(payload) +def _source_env(source_env: Mapping[str, str] | None = None) -> dict[str, str]: + if source_env is not None: + return dict(source_env) + resolved = read_repo_dotenv_values(_DOTENV_KEYS) + resolved.update(os.environ) + return resolved + + +def _normalise_user_concept_id(value: Any) -> str | None: + cleaned = clean_env_value(value if isinstance(value, str) else None) + if cleaned is None or not cleaned.startswith("#V#"): + return None + return cleaned + + +def _configured_resource_identity( + source_env: Mapping[str, str] | None = None, +) -> tuple[str | None, str | None]: + env = _source_env(source_env) + resource_id = ( + clean_env_value(env.get(LINKEDIN_RESOURCE_ID_ENV)) or DEFAULT_RESOURCE_ID + ) + owner_id = _normalise_user_concept_id( + env.get(LINKEDIN_OWNER_USER_CONCEPT_ID_ENV) + ) + return resource_id, owner_id + + +def linkedin_resource_binding_for_user( + user_concept_id: str | None, + *, + source_env: Mapping[str, str] | None = None, +) -> str | None: + """Return the private resource selector only for its configured owner. + + Callers must supply identity derived from Von's authenticated request + boundary. The selector, rather than a path or owner id, is injected into + the model tool call as a hidden trusted argument. + """ + + actor_id = _normalise_user_concept_id(user_concept_id) + resource_id, owner_id = _configured_resource_identity(source_env) + if actor_id is None or owner_id is None or actor_id != owner_id: + return None + return resource_id + + +def resolve_linkedin_invocation_authority( + *, + resource_id: str, + source_env: Mapping[str, str] | None = None, +) -> LinkedInInvocationAuthority: + """Recheck trusted invocation provenance and the owner binding.""" + + from .gateway import ( + get_internal_mcp_actor_context_source, + get_internal_mcp_preexisting_actor_context, + internal_mcp_actor_context_is_trusted_local_operator, + internal_mcp_actor_context_is_untrusted_payload_fallback, + ) + + configured_resource_id, owner_id = _configured_resource_identity(source_env) + requested_resource_id = str(resource_id or "").strip() + if owner_id is None or not configured_resource_id: + raise LinkedInInvocationAuthorityError( + reason_code="linkedin_owner_configuration_unavailable", + safe_message="LinkedIn resource ownership is not configured.", + ) + if requested_resource_id != configured_resource_id: + raise LinkedInInvocationAuthorityError( + reason_code="linkedin_resource_not_authorised", + safe_message=( + "The requested LinkedIn resource is not authorised for this invocation." + ), + ) + + if internal_mcp_actor_context_is_trusted_local_operator(): + return LinkedInInvocationAuthority( + principal_kind="trusted_local_operator", + resource_id=configured_resource_id, + ) + + if internal_mcp_actor_context_is_untrusted_payload_fallback(): + raise LinkedInInvocationAuthorityError( + reason_code="authenticated_actor_context_required", + safe_message=( + "LinkedIn access requires an authenticated actor or the trusted " + "local operator route; tool-payload identity is not authority." + ), + ) + + source = get_internal_mcp_actor_context_source() + preexisting_actor = get_internal_mcp_preexisting_actor_context() + actor_id = ( + _normalise_user_concept_id(preexisting_actor[0]) + if preexisting_actor is not None + else None + ) + if source is None or actor_id is None: + raise LinkedInInvocationAuthorityError( + reason_code="authenticated_actor_context_required", + safe_message="LinkedIn access requires an authenticated Von actor.", + ) + if actor_id != owner_id: + raise LinkedInInvocationAuthorityError( + reason_code="linkedin_resource_not_authorised", + safe_message="The requested LinkedIn resource is not authorised for this actor.", + ) + return LinkedInInvocationAuthority( + principal_kind="authenticated_actor", + actor_user_concept_id=actor_id, + resource_id=configured_resource_id, + ) class LinkedInMCPProxy: - """Manages local LinkedIn MCP server subprocess via shared MCP stdio client.""" + """Call a local LinkedIn MCP process bound to one configured database.""" def __init__(self, config: LinkedInProxyConfig): self._config = config @@ -109,95 +247,18 @@ def __init__(self, config: LinkedInProxyConfig): ) ) - async def _call(self, tool_name: str, arguments: dict[str, Any]) -> Any: + async def call(self, tool_name: str, arguments: dict[str, Any]) -> dict[str, Any]: try: - return await self._client.call_tool(tool_name, arguments) + payload = await self._client.call_tool(tool_name, arguments) except MCPToolClientError as exc: - raise LinkedInProxyError(str(exc)) from exc - - def _list_exports_from_data_root(self) -> list[str]: - root = self._config.data_root - if not root.exists() or not root.is_dir(): - return [] - - exports: list[str] = [] - for child in root.iterdir(): - name = child.name - if child.is_dir() or name.endswith(".zip") or name.endswith(".zip.zip"): - exports.append(name) - return sorted(exports, key=str.lower) - - async def list_exports(self, *, refresh: bool = False) -> dict[str, Any]: - if refresh: - logger.info("%s Refresh requested for local export listing", _LOG_TAG) - - exports = self._list_exports_from_data_root() - return { - "success": True, - "exports": exports, - "total_exports": len(exports), - "data_root": str(self._config.data_root), - "data_root_exists": self._config.data_root.exists(), - } - - async def list_files(self, *, export_name: str) -> dict[str, Any]: - raw = await self._call("list_files", {"export_name": export_name}) - return { - "success": True, - "export_name": export_name, - "files": _coerce_list_payload(raw), - } - - async def get_profile(self, *, export_name: str) -> dict[str, Any]: - raw = await self._call("get_profile", {"export_name": export_name}) - return { - "success": True, - "export_name": export_name, - "profile": _coerce_text_payload(raw), - } - - async def get_csv_data( - self, *, export_name: str, file_name: str, limit: int = 10 - ) -> dict[str, Any]: - raw = await self._call( - "get_csv_data", - {"export_name": export_name, "file_name": file_name, "limit": limit}, - ) - return { - "success": True, - "export_name": export_name, - "file_name": file_name, - "limit": limit, - "data": _coerce_text_payload(raw), - } - - async def get_company_stats( - self, *, export_name: str, top_n: int = 10 - ) -> dict[str, Any]: - raw = await self._call( - "get_company_stats", {"export_name": export_name, "top_n": top_n} - ) - stats_payload = _coerce_mapping_payload(raw) - company_stats = stats_payload.get("company_stats") - if not isinstance(company_stats, dict): - company_stats = stats_payload if isinstance(stats_payload, dict) else {} - return { - "success": True, - "export_name": export_name, - "top_n": top_n, - "company_stats": company_stats, - } - - async def get_messages(self, *, export_name: str, query: str = "") -> dict[str, Any]: - raw = await self._call( - "get_messages", {"export_name": export_name, "query": query} - ) - return { - "success": True, - "export_name": export_name, - "query": query, - "messages": _coerce_text_payload(raw), - } + raise LinkedInProxyError( + str(exc), + error_code="linkedin_mcp_call_failed", + details={"tool_name": tool_name}, + ) from exc + if isinstance(payload, dict): + return dict(payload) + return {"result": payload} def get_stats(self) -> dict[str, int]: return { @@ -206,84 +267,121 @@ def get_stats(self) -> dict[str, int]: } -def _compose_pythonpath(default_path: Path, existing: str | None) -> str: - parts = [str(default_path)] - if isinstance(existing, str) and existing.strip(): - parts.extend( - part.strip() - for part in existing.split(os.pathsep) - if isinstance(part, str) and part.strip() - ) - deduped: list[str] = [] - for part in parts: - if part not in deduped: - deduped.append(part) - return os.pathsep.join(deduped) - +def _default_entrypoint(project_dir: Path) -> Path: + if os.name == "nt": + return project_dir / ".venv" / "Scripts" / "linkedin-mjw-mcp.exe" + return project_dir / ".venv" / "bin" / "linkedin-mjw-mcp" -def _build_linkedin_config() -> LinkedInProxyConfig: - env = os.environ.copy() - server_path_raw = ( - os.environ.get("VON_LINKEDIN_MCP_SERVER_PATH") or str(_DEFAULT_SERVER_PATH) - ) - server_path = Path(server_path_raw).expanduser() - if not server_path.exists(): +def _build_linkedin_config( + *, + resource_id: str, + source_env: Mapping[str, str] | None = None, +) -> LinkedInProxyConfig: + env_source = _source_env(source_env) + configured_resource_id, owner_id = _configured_resource_identity(env_source) + if owner_id is None: raise LinkedInProxyError( - f"LinkedIn MCP server script not found at {server_path}. " - "Set VON_LINKEDIN_MCP_SERVER_PATH to the correct server.py path." + f"Set {LINKEDIN_OWNER_USER_CONCEPT_ID_ENV} to the owning Von user concept ID.", + error_code="linkedin_owner_configuration_unavailable", + ) + if resource_id != configured_resource_id: + raise LinkedInProxyError( + "Unknown LinkedIn resource selector.", + error_code="linkedin_resource_configuration_unavailable", ) - command = os.environ.get("VON_LINKEDIN_MCP_COMMAND", "python").strip() or "python" - args = [str(server_path)] + project_raw = clean_env_value(env_source.get(LINKEDIN_PROJECT_DIR_ENV)) + if project_raw is None: + raise LinkedInProxyError( + f"Set {LINKEDIN_PROJECT_DIR_ENV} to the separate LinkedInMJW checkout.", + error_code="linkedin_project_configuration_unavailable", + ) + project_dir = Path(project_raw).expanduser().resolve() + if not project_dir.is_dir(): + raise LinkedInProxyError( + f"LinkedIn MCP project directory does not exist: {project_dir}", + error_code="linkedin_project_unavailable", + ) - pythonpath_override = os.environ.get("VON_LINKEDIN_MCP_PYTHONPATH") - pythonpath_value = ( - pythonpath_override.strip() - if isinstance(pythonpath_override, str) and pythonpath_override.strip() - else str(_DEFAULT_PYTHONPATH) - ) - env["PYTHONPATH"] = _compose_pythonpath( - Path(pythonpath_value), env.get("PYTHONPATH") - ) + database_raw = clean_env_value(env_source.get(LINKEDIN_DATABASE_ENV)) + if database_raw is None: + raise LinkedInProxyError( + f"Set {LINKEDIN_DATABASE_ENV} to the private derived SQLite index.", + error_code="linkedin_database_configuration_unavailable", + ) + database = Path(database_raw).expanduser().resolve() + if not database.is_file(): + raise LinkedInProxyError( + f"LinkedIn index does not exist: {database}", + error_code="linkedin_database_unavailable", + ) - data_root_raw = os.environ.get("VON_LINKEDIN_DATA_ROOT") or str(_DEFAULT_DATA_ROOT) - data_root = Path(data_root_raw).expanduser() + command_override = clean_env_value(env_source.get(LINKEDIN_COMMAND_ENV)) + command = command_override or str(_default_entrypoint(project_dir)) + resolved_command = shutil.which(command) if not Path(command).is_absolute() else command + if not resolved_command or not Path(resolved_command).is_file(): + raise LinkedInProxyError( + f"LinkedIn MCP executable does not exist: {command}", + error_code="linkedin_executable_unavailable", + details={ + "recovery": ( + f"Run 'uv sync --frozen' in {project_dir} or set " + f"{LINKEDIN_COMMAND_ENV}." + ) + }, + ) - timeout_raw = os.environ.get("VON_LINKEDIN_MCP_TIMEOUT_SEC") + timeout_raw = clean_env_value(env_source.get(LINKEDIN_TIMEOUT_ENV)) try: timeout_sec = float(timeout_raw) if timeout_raw is not None else 30.0 - except Exception: + except ValueError: timeout_sec = 30.0 timeout_sec = max(5.0, min(300.0, timeout_sec)) + subprocess_env = { + key: value + for key in _SUBPROCESS_PASSTHROUGH_KEYS + if (value := env_source.get(key)) + } + subprocess_env[LINKEDIN_DATABASE_ENV] = str(database) return LinkedInProxyConfig( - command=command, - args=args, - env=env, - data_root=data_root, + resource_id=resource_id, + owner_user_concept_id=owner_id, + project_dir=project_dir, + database=database, + command=str(resolved_command), + args=[], + env=subprocess_env, timeout_sec=timeout_sec, ) -_proxy_instance: Optional[LinkedInMCPProxy] = None -# Catalogue calls can run on fresh event loops, so construction is loop-neutral. +_proxy_instance: LinkedInMCPProxy | None = None +_proxy_resource_id: str | None = None _proxy_lock = threading.Lock() -async def get_linkedin_proxy() -> LinkedInMCPProxy: - """Get or create the singleton LinkedIn MCP proxy instance.""" +async def get_linkedin_proxy(*, resource_id: str) -> LinkedInMCPProxy: + """Get the singleton proxy for the configured private resource.""" - global _proxy_instance + global _proxy_instance, _proxy_resource_id with _proxy_lock: - if _proxy_instance is None: - config = _build_linkedin_config() + if _proxy_instance is None or _proxy_resource_id != resource_id: + config = _build_linkedin_config(resource_id=resource_id) _proxy_instance = LinkedInMCPProxy(config) + _proxy_resource_id = resource_id logger.info( - "%s Initialised LinkedIn MCP proxy using %s %s", + "%s Initialised owner-scoped LinkedIn MCP proxy for resource=%s", _LOG_TAG, - config.command, - " ".join(config.args), + resource_id, ) return _proxy_instance + + +def reset_linkedin_proxy_for_tests() -> None: + global _proxy_instance, _proxy_resource_id + with _proxy_lock: + _proxy_instance = None + _proxy_resource_id = None diff --git a/src/backend/server/routes/von_routes.py b/src/backend/server/routes/von_routes.py index 176c8c8c..04c1e584 100644 --- a/src/backend/server/routes/von_routes.py +++ b/src/backend/server/routes/von_routes.py @@ -10436,6 +10436,19 @@ def _check_background_cancellation(subtask: str) -> None: _check_background_cancellation("authentication context") + linkedin_resource_trusted_binding: str | None = None + if user_concept_id: + from ...integrations.internal_mcp.linkedin_proxy_mcp import ( + linkedin_resource_binding_for_user, + ) + + # The body, conversation, and model cannot select a private LinkedIn + # resource. Only the authenticated request actor can acquire this + # hidden selector, and every handler rechecks the owner binding. + linkedin_resource_trusted_binding = linkedin_resource_binding_for_user( + user_concept_id + ) + authorised_gmail_profile: str | None = None gmail_profile_trusted_binding: Mapping[str, Any] | str | None = None if not user_concept_id and request_gmail_profile: @@ -11610,6 +11623,10 @@ def _progress_update(info: dict[str, Any]) -> None: adaptive_turn_started = time.perf_counter() adaptive_input_context = enhanced_context trusted_turn_argument_values: dict[str, Any] = {} + if linkedin_resource_trusted_binding is not None: + trusted_turn_argument_values["linkedin_resource_id"] = ( + linkedin_resource_trusted_binding + ) if gmail_profile_trusted_binding is not None: trusted_turn_argument_values["gmail_profile"] = ( gmail_profile_trusted_binding diff --git a/src/backend/services/tool_metadata_service.py b/src/backend/services/tool_metadata_service.py index ebcb09d5..0a2313f4 100644 --- a/src/backend/services/tool_metadata_service.py +++ b/src/backend/services/tool_metadata_service.py @@ -459,36 +459,41 @@ class ToolRequiredObligationMetadata: "category": "arxiv", "display_template": "Materialised paper: {paper_concept_id}", }, - # LinkedIn Data Dump tools - "linkedin_list_exports": { - "salience": "high", + # Owner-scoped LinkedIn export tools + "linkedin_index_status": { + "salience": "medium", + "category": "linkedin", + "display_template": "LinkedIn index: {records} records", + }, + "linkedin_list_datasets": { + "salience": "medium", "category": "linkedin", - "display_template": "{total_exports} exports", + "display_template": "LinkedIn datasets: {count}", }, - "linkedin_list_files": { + "linkedin_search_export": { "salience": "high", "category": "linkedin", - "display_template": "Files: {export_name}", + "display_template": "LinkedIn export matches: {count}", }, - "linkedin_get_profile": { - "salience": "medium", + "linkedin_search_connections": { + "salience": "high", "category": "linkedin", - "display_template": "Profile: {export_name}", + "display_template": "LinkedIn connections: {count}", }, - "linkedin_get_csv_data": { - "salience": "medium", + "linkedin_search_messages": { + "salience": "high", "category": "linkedin", - "display_template": "CSV: {file_name}", + "display_template": "LinkedIn messages: {count}", }, - "linkedin_get_company_stats": { + "linkedin_list_connection_organisations": { "salience": "medium", "category": "linkedin", - "display_template": "Company stats: {export_name}", + "display_template": "LinkedIn organisations: {count}", }, - "linkedin_get_messages": { + "linkedin_get_record": { "salience": "medium", "category": "linkedin", - "display_template": "Messages: {export_name}", + "display_template": "LinkedIn record: {record_id}", }, # Web/search tools (HIGH salience) "search_web": { diff --git a/src/workflows/von/main.py b/src/workflows/von/main.py index 4da4bbde..07fe368a 100644 --- a/src/workflows/von/main.py +++ b/src/workflows/von/main.py @@ -131,6 +131,12 @@ def _apply_dotenv_overrides(keys: set[str]) -> None: "TAVILY_API_KEY", "VON_GMAIL_PROFILES", "VON_GMAIL_DEFAULT_PROFILE", + "VON_LINKEDIN_RESOURCE_ID", + "VON_LINKEDIN_OWNER_USER_CONCEPT_ID", + "VON_LINKEDIN_MCP_PROJECT_DIR", + "VON_LINKEDIN_MCP_COMMAND", + "VON_LINKEDIN_MCP_TIMEOUT_SEC", + "LINKEDIN_MJW_DB", "VON_INTERNAL_MCP_ENABLE", "VON_MCP_ALLOW_WRITES", "VON_INTERNAL_MCP_ALLOW_USER_TOOL_CALLS", diff --git a/tests/backend/test_internal_mcp_catalogue_builds.py b/tests/backend/test_internal_mcp_catalogue_builds.py index af799a9a..146fde9c 100644 --- a/tests/backend/test_internal_mcp_catalogue_builds.py +++ b/tests/backend/test_internal_mcp_catalogue_builds.py @@ -588,7 +588,7 @@ def test_ordinary_turn_read_projection_follows_capability_authority_metadata(): "github_list_tools", "gmail_list_profiles", "jira_search", - "linkedin_get_messages", + "linkedin_search_messages", "list_recent_screenshots", "mongo_cost_guardrails_report", "mongo_query_diagnostics_report", diff --git a/tests/backend/test_internal_mcp_linkedin_tools.py b/tests/backend/test_internal_mcp_linkedin_tools.py index 0a5e748c..be793461 100644 --- a/tests/backend/test_internal_mcp_linkedin_tools.py +++ b/tests/backend/test_internal_mcp_linkedin_tools.py @@ -1,138 +1,238 @@ -"""Smoke tests for LinkedIn MCP integration in the internal MCP catalogue.""" - -from src.backend.integrations.internal_mcp.catalogue import ( - _linkedin_get_company_stats, - _linkedin_get_csv_data, - _linkedin_get_messages, - _linkedin_get_profile, - _linkedin_list_files, - build_default_catalogue, +"""Owner-scope and protocol-adapter tests for private LinkedIn retrieval.""" + +from pathlib import Path + +from src.backend.integrations.internal_mcp.catalogue import build_default_catalogue +from src.backend.integrations.internal_mcp.gateway import InternalMCPGateway +from src.backend.integrations.internal_mcp.linkedin_proxy_mcp import ( + LINKEDIN_DATABASE_ENV, + LINKEDIN_OWNER_USER_CONCEPT_ID_ENV, + LINKEDIN_PROJECT_DIR_ENV, + _build_linkedin_config, + linkedin_resource_binding_for_user, +) +from src.backend.integrations.internal_mcp.transport import InternalMCPTransport +from src.backend.security.access_control import override_current_actor +from src.backend.services.adaptive_turn_service import ( + _model_visible_input_schema, + ordinary_turn_capability_delegation, ) +OWNER_ID = "#V#linkedin_owner" +OTHER_ID = "#V#other_user" +RESOURCE_ID = "personal_linkedin" -def test_linkedin_methods_registered_in_catalogue(): - catalogue = build_default_catalogue() - names = set(catalogue.list_methods()) - assert "linkedin_list_exports" in names - assert "linkedin_list_files" in names - assert "linkedin_get_profile" in names - assert "linkedin_get_csv_data" in names - assert "linkedin_get_company_stats" in names - assert "linkedin_get_messages" in names +def _configure_owner(monkeypatch) -> None: + monkeypatch.setenv(LINKEDIN_OWNER_USER_CONCEPT_ID_ENV, OWNER_ID) -def test_linkedin_handlers_require_minimum_fields(): - err_files = _linkedin_list_files(export_name=None) - assert err_files.get("success") is False - assert "export_name" in err_files.get("error", "") +def test_new_linkedin_search_methods_replace_legacy_surface(): + names = set(build_default_catalogue().list_methods()) - err_profile = _linkedin_get_profile(export_name=None) - assert err_profile.get("success") is False - assert "export_name" in err_profile.get("error", "") + assert { + "linkedin_index_status", + "linkedin_list_datasets", + "linkedin_search_export", + "linkedin_search_connections", + "linkedin_search_messages", + "linkedin_list_connection_organisations", + "linkedin_get_record", + } <= names + assert { + "linkedin_list_exports", + "linkedin_list_files", + "linkedin_get_profile", + "linkedin_get_csv_data", + "linkedin_get_company_stats", + "linkedin_get_messages", + }.isdisjoint(names) - err_csv = _linkedin_get_csv_data(export_name=None, file_name=None) - assert err_csv.get("success") is False - assert "export_name" in err_csv.get("error", "") - assert "file_name" in err_csv.get("error", "") - err_company = _linkedin_get_company_stats(export_name=None) - assert err_company.get("success") is False - assert "export_name" in err_company.get("error", "") +def test_resource_binding_is_returned_only_for_configured_owner(): + env = {LINKEDIN_OWNER_USER_CONCEPT_ID_ENV: OWNER_ID} - err_messages = _linkedin_get_messages(export_name=None) - assert err_messages.get("success") is False - assert "export_name" in err_messages.get("error", "") + assert linkedin_resource_binding_for_user(OWNER_ID, source_env=env) == RESOURCE_ID + assert linkedin_resource_binding_for_user(OTHER_ID, source_env=env) is None + assert linkedin_resource_binding_for_user(None, source_env=env) is None -def test_linkedin_get_csv_data_rejects_invalid_limit(): - result = _linkedin_get_csv_data( - export_name="sample-export", - file_name="Profile.csv", - limit="ten", +def test_linkedin_tools_are_projected_only_with_server_bound_resource(): + gateway = InternalMCPGateway( + catalogue=build_default_catalogue(), + transport=InternalMCPTransport(), + enabled=True, + ) + + without_binding = ordinary_turn_capability_delegation( + gateway, + user_concept_id=OWNER_ID, + trusted_argument_values={}, + ) + with_binding = ordinary_turn_capability_delegation( + gateway, + user_concept_id=OWNER_ID, + trusted_argument_values={"linkedin_resource_id": RESOURCE_ID}, ) - assert result.get("success") is False - assert result.get("error_code") == "invalid_parameter" - - -def test_linkedin_tools_success_through_gateway_invoke(monkeypatch): - from src.backend.integrations.internal_mcp.gateway import InternalMCPGateway - from src.backend.integrations.internal_mcp.transport import InternalMCPTransport - - class _FakeProxy: - async def list_exports(self, *, refresh: bool = False): - assert refresh is False - return { - "success": True, - "exports": ["export_a.zip"], - "total_exports": 1, - "data_root": "data/linkedin", - "data_root_exists": True, - } - - async def list_files(self, *, export_name: str): - assert export_name == "export_a.zip" - return { - "success": True, - "export_name": export_name, - "files": ["Profile.csv", "Connections.csv"], - } - - async def _fake_get_linkedin_proxy(): - return _FakeProxy() + + assert "linkedin_search_connections" not in without_binding + assert "linkedin_search_connections" in with_binding + + definition = gateway.get_method_definition("linkedin_search_connections") + assert definition is not None + visible_schema = _model_visible_input_schema( + definition, + {"linkedin_resource_id": RESOURCE_ID}, + ) + assert "resource_id" not in visible_schema["properties"] + + +def test_proxy_config_uses_one_database_and_minimal_environment(tmp_path: Path): + project = tmp_path / "LinkedInMJW" + executable = project / ".venv" / "bin" / "linkedin-mjw-mcp" + executable.parent.mkdir(parents=True) + executable.write_text("#!/bin/sh\n", encoding="utf-8") + database = tmp_path / "linkedin.sqlite3" + database.write_bytes(b"sqlite-placeholder") + env = { + LINKEDIN_OWNER_USER_CONCEPT_ID_ENV: OWNER_ID, + LINKEDIN_PROJECT_DIR_ENV: str(project), + LINKEDIN_DATABASE_ENV: str(database), + "PATH": "/usr/bin", + "OPENAI_API_KEY": "must-not-cross-boundary", + } + + config = _build_linkedin_config(resource_id=RESOURCE_ID, source_env=env) + + assert config.database == database.resolve() + assert config.command == str(executable.resolve()) + assert config.args == [] + assert config.env[LINKEDIN_DATABASE_ENV] == str(database.resolve()) + assert "OPENAI_API_KEY" not in config.env + + +def test_authenticated_owner_can_call_and_receives_authority_receipt(monkeypatch): + _configure_owner(monkeypatch) + + class FakeProxy: + async def call(self, tool_name, arguments): + assert tool_name == "search_connections" + assert arguments == {"company": "DeepMind", "limit": 5} + return {"count": 1, "results": [{"person": "Ada"}]} + + async def fake_get_linkedin_proxy(*, resource_id): + assert resource_id == RESOURCE_ID + return FakeProxy() monkeypatch.setattr( "src.backend.integrations.internal_mcp.linkedin_proxy_mcp.get_linkedin_proxy", - _fake_get_linkedin_proxy, + fake_get_linkedin_proxy, ) - gateway = InternalMCPGateway( catalogue=build_default_catalogue(), transport=InternalMCPTransport(), enabled=True, ) - exports_result = gateway.invoke("linkedin_list_exports", {}) - exports_payload = exports_result.payload - assert exports_payload.get("success") is True - assert exports_payload.get("total_exports") == 1 + with override_current_actor(OWNER_ID, None): + result = gateway.invoke( + "linkedin_search_connections", + { + "resource_id": RESOURCE_ID, + "company": "DeepMind", + "limit": 5, + }, + ).payload + + assert result["success"] is True + assert result["count"] == 1 + assert result["authority"] == { + "schema_version": "linkedin_resource_authority.v1", + "authorised": True, + "principal_kind": "authenticated_actor", + "actor_user_concept_id": OWNER_ID, + "resource_id": RESOURCE_ID, + "grant_source": "deployment_owner_binding", + "access": "read_only", + } + + +def test_other_actor_is_denied_before_proxy_launch(monkeypatch): + _configure_owner(monkeypatch) + proxy_called = False + + async def fake_get_linkedin_proxy(*, resource_id): + nonlocal proxy_called + proxy_called = True + raise AssertionError("private subprocess must not start") - files_result = gateway.invoke( - "linkedin_list_files", - {"export_name": "export_a.zip"}, + monkeypatch.setattr( + "src.backend.integrations.internal_mcp.linkedin_proxy_mcp.get_linkedin_proxy", + fake_get_linkedin_proxy, + ) + gateway = InternalMCPGateway( + catalogue=build_default_catalogue(), + transport=InternalMCPTransport(), + enabled=True, ) - files_payload = files_result.payload - assert files_payload.get("success") is True - assert files_payload.get("export_name") == "export_a.zip" - assert files_payload.get("files") == ["Profile.csv", "Connections.csv"] + with override_current_actor(OTHER_ID, None): + result = gateway.invoke( + "linkedin_index_status", + {"resource_id": RESOURCE_ID}, + ).payload -def test_linkedin_list_files_proxy_error_through_gateway_invoke(monkeypatch): - from src.backend.integrations.internal_mcp.gateway import InternalMCPGateway - from src.backend.integrations.internal_mcp.transport import InternalMCPTransport - from src.backend.integrations.internal_mcp.linkedin_proxy_mcp import ( - LinkedInProxyError, + assert result["success"] is False + assert result["error_code"] == "linkedin_resource_not_authorised" + assert proxy_called is False + + +def test_payload_identity_is_not_authority(monkeypatch): + _configure_owner(monkeypatch) + gateway = InternalMCPGateway( + catalogue=build_default_catalogue(), + transport=InternalMCPTransport(), + enabled=True, ) - class _FailingProxy: - async def list_files(self, *, export_name: str): # noqa: ARG002 - raise LinkedInProxyError("proxy unavailable") + result = gateway.invoke( + "linkedin_index_status", + {"resource_id": RESOURCE_ID}, + ).payload + + assert result["success"] is False + assert result["error_code"] == "authenticated_actor_context_required" + - async def _fake_get_linkedin_proxy(): - return _FailingProxy() +def test_trusted_local_operator_is_distinguished_from_google_actor(monkeypatch): + _configure_owner(monkeypatch) + + class FakeProxy: + async def call(self, tool_name, arguments): + assert tool_name == "index_status" + assert arguments == {} + return {"ready": True, "records": 7} + + async def fake_get_linkedin_proxy(*, resource_id): + assert resource_id == RESOURCE_ID + return FakeProxy() monkeypatch.setattr( "src.backend.integrations.internal_mcp.linkedin_proxy_mcp.get_linkedin_proxy", - _fake_get_linkedin_proxy, + fake_get_linkedin_proxy, ) - gateway = InternalMCPGateway( catalogue=build_default_catalogue(), transport=InternalMCPTransport(), enabled=True, + trusted_actor_payload_fallback=True, ) - result = gateway.invoke("linkedin_list_files", {"export_name": "export_a.zip"}) - payload = result.payload - assert payload.get("success") is False - assert payload.get("error_code") == "linkedin_proxy_error" + result = gateway.invoke( + "linkedin_index_status", + {"resource_id": RESOURCE_ID}, + ).payload + + assert result["success"] is True + assert result["authority"]["principal_kind"] == "trusted_local_operator" + assert result["authority"]["actor_user_concept_id"] is None