Coverage for agentos/marketplace/bridge.py: 0%
229 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-06 10:59 +0800
1"""Skill marketplace ecosystem bridge.
3Converts skills from external ecosystems (Claude Code, Cursor, Custom GPT, LangChain)
4into AgentOS SkillManifest format for unified skill registry and discovery.
5"""
7from __future__ import annotations
9import enum
10import logging
11import os
12import tempfile
13from dataclasses import dataclass, field
14from pathlib import Path
15from typing import Dict, List, Optional
17from agentos.marketplace.manifest import SkillManifest, SkillFormat, ToolDef
19logger = logging.getLogger(__name__)
22# ── Ecosystem Formats ───────────────────────────────────────────────
25class EcosystemFormat(str, enum.Enum):
26 """Supported external ecosystem formats."""
27 CLAUDE_CODE = "claude-code"
28 CURSOR = "cursor"
29 CUSTOM_GPT = "custom-gpt"
30 LANGCHAIN = "langchain"
33# ── Data Classes ────────────────────────────────────────────────────
36@dataclass
37class BridgeResult:
38 """Result of bridging a single skill from an external ecosystem."""
39 success: bool = False
40 skill_name: str = ""
41 source_format: str = ""
42 source_uri: str = ""
43 manifest: Optional[SkillManifest] = None
44 error: str = ""
45 warnings: List[str] = field(default_factory=list)
48@dataclass
49class BridgeBatchResult:
50 """Result of a batch bridge operation."""
51 total: int = 0
52 succeeded: int = 0
53 failed: int = 0
54 results: List[BridgeResult] = field(default_factory=list)
55 errors: List[str] = field(default_factory=list)
58# ── Base Adapter ────────────────────────────────────────────────────
61class BaseAdapter:
62 """Base class for ecosystem adapters."""
64 format_name: str = ""
66 def detect(self, source: str) -> bool:
67 """Check if this adapter can handle the given source."""
68 raise NotImplementedError
70 def bridge(self, source: str) -> BridgeResult:
71 """Bridge a single skill from external format to AgentOS."""
72 raise NotImplementedError
74 def list_available(self) -> List[str]:
75 """List available skills in this ecosystem."""
76 return []
79# ── Claude Code Adapter ────────────────────────────────────────────
82class ClaudeCodeAdapter(BaseAdapter):
83 """Bridge Claude Code extensions to AgentOS skills.
85 Claude Code extensions are npm packages that expose tools or MCP servers.
86 This adapter can:
87 1. Download the package from npm (or read local)
88 2. Parse the package.json and extension manifest
89 3. Convert tool definitions to AgentOS ToolDef
90 4. Generate a SkillManifest
91 """
93 format_name = EcosystemFormat.CLAUDE_CODE.value
95 def __init__(self, cache_dir: Optional[str] = None):
96 self._cache_dir = cache_dir or os.path.join(
97 tempfile.gettempdir(), "agentos", "claude_cache"
98 )
99 os.makedirs(self._cache_dir, exist_ok=True)
101 def detect(self, source: str) -> bool:
102 return (
103 source.startswith("claude://")
104 or source.startswith("@") # npm scoped package
105 or "claude-code" in source.lower()
106 or source.endswith(".tgz")
107 )
109 def list_available(self) -> List[str]:
110 """Return known popular Claude Code extensions."""
111 return [
112 "@anthropic/claude-code-tools",
113 "@modelcontextprotocol/server-filesystem",
114 "@modelcontextprotocol/server-github",
115 "@modelcontextprotocol/server-postgres",
116 "@modelcontextprotocol/server-sqlite",
117 "@modelcontextprotocol/server-puppeteer",
118 "@modelcontextprotocol/server-playwright",
119 "@modelcontextprotocol/server-redis",
120 ]
122 def bridge(self, source: str) -> BridgeResult:
123 result = BridgeResult(
124 skill_name=source,
125 source_format=self.format_name,
126 source_uri=source,
127 )
129 try:
130 # Strip protocol prefix
131 if source.startswith("claude://"):
132 source = source[len("claude://"):]
134 # Try to load extension manifest (simulated for now)
135 manifest = self._convert_to_skill(source)
136 if manifest:
137 result.success = True
138 result.manifest = manifest
139 result.skill_name = manifest.name
140 result.warnings.append("Claude Code extension converted to AgentOS format")
141 result.warnings.append(
142 "Note: Some Claude Code extensions use external APIs "
143 "that may require additional configuration"
144 )
145 else:
146 result.error = f"Could not parse Claude Code extension: {source}"
148 except Exception as e:
149 result.error = f"Bridge failed: {e}"
151 return result
153 def _convert_to_skill(self, source: str) -> Optional[SkillManifest]:
154 """Convert a Claude Code extension identifier to a SkillManifest.
156 In production, this would:
157 1. Download the npm package
158 2. Parse package.json for 'claude-code' extension config
159 3. Convert tool definitions
161 For now, we generate a template manifest based on the package name.
162 """
163 name = source.lstrip("@").replace("/", "-").replace("@", "")
164 # Infer tools from package name
165 tools = []
166 if "filesystem" in source.lower():
167 tools.append(ToolDef(
168 name="read_file",
169 description="Read file contents from the filesystem",
170 parameters={"type": "object", "properties": {
171 "path": {"type": "string"},
172 }},
173 ))
174 tools.append(ToolDef(
175 name="write_file",
176 description="Write content to a file",
177 parameters={"type": "object", "properties": {
178 "path": {"type": "string"},
179 "content": {"type": "string"},
180 }},
181 ))
182 elif "github" in source.lower():
183 tools.append(ToolDef(
184 name="github_get_file",
185 description="Get file contents from a GitHub repository",
186 parameters={"type": "object", "properties": {
187 "owner": {"type": "string"},
188 "repo": {"type": "string"},
189 "path": {"type": "string"},
190 }},
191 ))
192 elif "database" in source.lower() or "postgres" in source.lower():
193 tools.append(ToolDef(
194 name="query_database",
195 description="Execute a SQL query against the database",
196 parameters={"type": "object", "properties": {
197 "query": {"type": "string"},
198 }},
199 ))
201 return SkillManifest(
202 name=name,
203 version="1.0.0",
204 description=f"Claude Code extension: {source}",
205 format=SkillFormat.GENERIC,
206 tools=tools if tools else [
207 ToolDef(
208 name=f"{name}_tool",
209 description=f"Auto-converted tool from {source}",
210 parameters={"type": "object", "properties": {}},
211 )
212 ],
213 author="Claude Code Ecosystem",
214 tags=["claude-code", "bridge"],
215 )
218# ── Cursor Adapter ──────────────────────────────────────────────────
221class CursorAdapter(BaseAdapter):
222 """Bridge Cursor rules to AgentOS skills.
224 Cursor uses .cursorrules files and .cursor/rules/ directories
225 to define AI behavior modifications. This adapter converts
226 those rule definitions into AgentOS skills.
227 """
229 format_name = EcosystemFormat.CURSOR.value
231 def detect(self, source: str) -> bool:
232 return (
233 source.startswith("cursor://")
234 or ".cursorrules" in source.lower()
235 or ".cursor/" in source
236 or source.endswith(".mdc")
237 )
239 def list_available(self) -> List[str]:
240 """Return common Cursor rule sources."""
241 return [
242 "cursor://rules/python-best-practices",
243 "cursor://rules/typescript-standards",
244 "cursor://rules/react-patterns",
245 "cursor://rules/testing-guidelines",
246 ]
248 def bridge(self, source: str) -> BridgeResult:
249 result = BridgeResult(
250 skill_name=source,
251 source_format=self.format_name,
252 source_uri=source,
253 )
255 try:
256 if source.startswith("cursor://"):
257 rule_path = source[len("cursor://"):]
258 else:
259 rule_path = source
261 manifest = self._convert_rule(rule_path)
262 if manifest:
263 result.success = True
264 result.manifest = manifest
265 result.skill_name = manifest.name
266 result.warnings.append("Cursor rule converted to AgentOS skill")
267 else:
268 result.error = f"Could not parse Cursor rule: {source}"
270 except Exception as e:
271 result.error = f"Bridge failed: {e}"
273 return result
275 def _convert_rule(self, rule_path: str) -> Optional[SkillManifest]:
276 name = Path(rule_path).stem.replace(".cursorrules", "").replace(".", "-")
277 if not name:
278 name = rule_path.replace("/", "-")
280 return SkillManifest(
281 name=name,
282 version="1.0.0",
283 description=f"Cursor rule: {rule_path}",
284 format=SkillFormat.GENERIC,
285 tools=[],
286 author="Cursor Ecosystem",
287 tags=["cursor", "bridge"],
288 )
291# ── Custom GPT Adapter ──────────────────────────────────────────────
294class CustomGPTAdapter(BaseAdapter):
295 """Bridge Custom GPT instructions to AgentOS skills.
297 Custom GPTs have instructions, conversation starters, knowledge files,
298 and capabilities. This adapter extracts instructions and converts
299 them into an AgentOS skill definition.
300 """
302 format_name = EcosystemFormat.CUSTOM_GPT.value
304 def detect(self, source: str) -> bool:
305 return (
306 source.startswith("gpt://")
307 or "chatgpt.com/g/" in source
308 or source.endswith(".gpt.md")
309 )
311 def list_available(self) -> List[str]:
312 return [
313 "gpt://data-analyst",
314 "gpt://creative-writer",
315 "gpt://code-reviewer",
316 "gpt://research-assistant",
317 ]
319 def bridge(self, source: str) -> BridgeResult:
320 result = BridgeResult(
321 skill_name=source,
322 source_format=self.format_name,
323 source_uri=source,
324 )
326 try:
327 if source.startswith("gpt://"):
328 gpt_id = source[len("gpt://"):]
329 else:
330 gpt_id = source
332 manifest = self._convert_gpt(gpt_id)
333 if manifest:
334 result.success = True
335 result.manifest = manifest
336 result.skill_name = manifest.name
337 result.warnings.append("Custom GPT instructions converted to AgentOS skill")
338 else:
339 result.error = f"Could not parse Custom GPT: {source}"
341 except Exception as e:
342 result.error = f"Bridge failed: {e}"
344 return result
346 def _convert_gpt(self, gpt_id: str) -> Optional[SkillManifest]:
347 name = gpt_id.replace("/", "-").replace(" ", "-")
348 return SkillManifest(
349 name=name,
350 version="1.0.0",
351 description=f"Custom GPT: {gpt_id}",
352 format=SkillFormat.GENERIC,
353 tools=[],
354 author="Custom GPT Ecosystem",
355 tags=["custom-gpt", "bridge"],
356 )
359# ── LangChain Adapter ───────────────────────────────────────────────
362class LangChainAdapter(BaseAdapter):
363 """Bridge LangChain tools to AgentOS skills.
365 LangChain provides a rich ecosystem of tools (toolkits, tools,
366 MCP adapters). This adapter converts them into AgentOS ToolDef
367 and wraps them in a SkillManifest.
368 """
370 format_name = EcosystemFormat.LANGCHAIN.value
372 KNOWN_TOOLS = {
373 "wikipedia": {
374 "name": "wikipedia_query",
375 "description": "Search and retrieve information from Wikipedia",
376 "parameters": {
377 "type": "object",
378 "properties": {
379 "query": {"type": "string", "description": "Search query"},
380 "max_results": {"type": "integer", "default": 3},
381 },
382 "required": ["query"],
383 },
384 },
385 "arxiv": {
386 "name": "arxiv_search",
387 "description": "Search academic papers on arXiv",
388 "parameters": {
389 "type": "object",
390 "properties": {
391 "query": {"type": "string"},
392 "max_results": {"type": "integer", "default": 5},
393 },
394 "required": ["query"],
395 },
396 },
397 "duckduckgo": {
398 "name": "web_search",
399 "description": "Search the web using DuckDuckGo",
400 "parameters": {
401 "type": "object",
402 "properties": {
403 "query": {"type": "string"},
404 },
405 "required": ["query"],
406 },
407 },
408 "python_repl": {
409 "name": "execute_python",
410 "description": "Execute Python code in a REPL environment",
411 "parameters": {
412 "type": "object",
413 "properties": {
414 "code": {"type": "string"},
415 },
416 "required": ["code"],
417 },
418 },
419 "shell": {
420 "name": "execute_shell",
421 "description": "Execute shell commands",
422 "parameters": {
423 "type": "object",
424 "properties": {
425 "command": {"type": "string"},
426 },
427 "required": ["command"],
428 },
429 },
430 }
432 def detect(self, source: str) -> bool:
433 return (
434 source.startswith("langchain://")
435 or "langchain" in source.lower()
436 )
438 def list_available(self) -> List[str]:
439 return [f"langchain://{name}" for name in self.KNOWN_TOOLS]
441 def bridge(self, source: str) -> BridgeResult:
442 result = BridgeResult(
443 skill_name=source,
444 source_format=self.format_name,
445 source_uri=source,
446 )
448 try:
449 if source.startswith("langchain://"):
450 tool_name = source[len("langchain://"):]
451 else:
452 tool_name = source
454 manifest = self._convert_tool(tool_name)
455 if manifest:
456 result.success = True
457 result.manifest = manifest
458 result.skill_name = manifest.name
459 else:
460 result.error = f"Unknown LangChain tool: {tool_name}"
462 except Exception as e:
463 result.error = f"Bridge failed: {e}"
465 return result
467 def _convert_tool(self, tool_name: str) -> Optional[SkillManifest]:
468 if tool_name not in self.KNOWN_TOOLS:
469 return None
471 tool_info = self.KNOWN_TOOLS[tool_name]
472 tool_def = ToolDef(
473 name=tool_info["name"],
474 description=tool_info["description"],
475 parameters=tool_info["parameters"],
476 )
478 return SkillManifest(
479 name=f"langchain-{tool_name}",
480 version="1.0.0",
481 description=f"LangChain tool: {tool_name}",
482 format=SkillFormat.GENERIC,
483 tools=[tool_def],
484 author="LangChain Ecosystem",
485 tags=["langchain", "bridge"],
486 )
489# ── Adapter Factory ─────────────────────────────────────────────────
492class AdapterFactory:
493 """Factory for creating ecosystem adapters."""
495 _adapters: Dict[EcosystemFormat, type] = {}
497 @classmethod
498 def register(cls, fmt: EcosystemFormat, adapter_cls: type):
499 cls._adapters[fmt] = adapter_cls
501 @classmethod
502 def create(cls, fmt: EcosystemFormat, **kwargs) -> BaseAdapter:
503 """Create an adapter for the given ecosystem format."""
504 if fmt not in cls._adapters:
505 raise ValueError(f"Unsupported ecosystem format: {fmt}")
506 return cls._adapters[fmt](**kwargs)
508 @classmethod
509 def detect_format(cls, source: str) -> Optional[EcosystemFormat]:
510 """Auto-detect ecosystem format from source string."""
511 for fmt, adapter_cls in cls._adapters.items():
512 adapter = adapter_cls()
513 if adapter.detect(source):
514 return fmt
515 return None
517 @classmethod
518 def list_supported_formats(cls) -> List[str]:
519 return [f.value for f in cls._adapters]
522# Register built-in adapters
523AdapterFactory.register(EcosystemFormat.CLAUDE_CODE, ClaudeCodeAdapter)
524AdapterFactory.register(EcosystemFormat.CURSOR, CursorAdapter)
525AdapterFactory.register(EcosystemFormat.CUSTOM_GPT, CustomGPTAdapter)
526AdapterFactory.register(EcosystemFormat.LANGCHAIN, LangChainAdapter)
529# ── Ecosystem Bridge (Main Entry) ───────────────────────────────────
532class EcosystemBridge:
533 """Bridge external skill ecosystems into AgentOS SkillRegistry.
535 Usage:
536 bridge = EcosystemBridge()
537 # Single skill
538 result = bridge.bridge("claude://@anthropic/claude-code-tools")
540 # Batch (all available from one ecosystem)
541 results = bridge.bridge_all(EcosystemFormat.CLAUDE_CODE)
543 # Auto-detect and bridge
544 result = bridge.bridge("langchain://wikipedia")
545 """
547 def __init__(self, skill_registry=None):
548 self._skill_registry = skill_registry
549 self._adapters: Dict[EcosystemFormat, BaseAdapter] = {}
551 def _get_adapter(self, fmt: EcosystemFormat) -> BaseAdapter:
552 if fmt not in self._adapters:
553 self._adapters[fmt] = AdapterFactory.create(fmt)
554 return self._adapters[fmt]
556 def bridge(self, source: str, fmt: Optional[EcosystemFormat] = None) -> BridgeResult:
557 """Bridge a single skill from external ecosystem.
559 Args:
560 source: Source identifier (e.g., "claude://pkg", "cursor://rule")
561 fmt: Ecosystem format. Auto-detected if not specified.
563 Returns:
564 BridgeResult with converted SkillManifest.
565 """
566 if not fmt:
567 fmt = AdapterFactory.detect_format(source)
568 if not fmt:
569 return BridgeResult(
570 success=False,
571 skill_name=source,
572 error=f"Could not auto-detect ecosystem format for: {source}",
573 )
575 adapter = self._get_adapter(fmt)
576 result = adapter.bridge(source)
578 # Auto-register to skill registry if available
579 if result.success and result.manifest and self._skill_registry:
580 try:
581 self._skill_registry.register_skill(result.manifest)
582 except Exception as e:
583 result.warnings.append(f"Registered but SKillRegistry error: {e}")
585 return result
587 def bridge_all(self, fmt: EcosystemFormat) -> BridgeBatchResult:
588 """Bridge all available skills from an ecosystem."""
589 batch = BridgeBatchResult()
590 adapter = self._get_adapter(fmt)
591 available = adapter.list_available()
593 for source in available:
594 result = adapter.bridge(source)
595 batch.results.append(result)
596 if result.success:
597 batch.succeeded += 1
598 else:
599 batch.failed += 1
600 batch.errors.append(result.error)
602 batch.total = len(available)
603 return batch
605 def batch_bridge(self, sources: List[str]) -> BridgeBatchResult:
606 """Bridge multiple sources, auto-detecting formats."""
607 batch = BridgeBatchResult()
609 for source in sources:
610 result = self.bridge(source)
611 batch.results.append(result)
612 if result.success:
613 batch.succeeded += 1
614 else:
615 batch.failed += 1
616 batch.errors.append(result.error)
618 batch.total = len(sources)
619 return batch
621 def list_available(self, fmt: EcosystemFormat) -> List[str]:
622 """List available skills in an ecosystem."""
623 adapter = self._get_adapter(fmt)
624 return adapter.list_available()
626 def supported_formats(self) -> List[str]:
627 return AdapterFactory.list_supported_formats()