Coverage for agentos/marketplace/bridge.py: 0%
228 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-06 19:15 +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
16from agentos.marketplace.manifest import SkillFormat, SkillManifest, ToolDef
18logger = logging.getLogger(__name__)
21# ── Ecosystem Formats ───────────────────────────────────────────────
24class EcosystemFormat(enum.StrEnum):
25 """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."""
40 success: bool = False
41 skill_name: str = ""
42 source_format: str = ""
43 source_uri: str = ""
44 manifest: SkillManifest | None = None
45 error: str = ""
46 warnings: list[str] = field(default_factory=list)
49@dataclass
50class BridgeBatchResult:
51 """Result of a batch bridge operation."""
53 total: int = 0
54 succeeded: int = 0
55 failed: int = 0
56 results: list[BridgeResult] = field(default_factory=list)
57 errors: list[str] = field(default_factory=list)
60# ── Base Adapter ────────────────────────────────────────────────────
63class BaseAdapter:
64 """Base class for ecosystem adapters."""
66 format_name: str = ""
68 def detect(self, source: str) -> bool:
69 """Check if this adapter can handle the given source."""
70 raise NotImplementedError
72 def bridge(self, source: str) -> BridgeResult:
73 """Bridge a single skill from external format to AgentOS."""
74 raise NotImplementedError
76 def list_available(self) -> list[str]:
77 """List available skills in this ecosystem."""
78 return []
81# ── Claude Code Adapter ────────────────────────────────────────────
84class ClaudeCodeAdapter(BaseAdapter):
85 """Bridge Claude Code extensions to AgentOS skills.
87 Claude Code extensions are npm packages that expose tools or MCP servers.
88 This adapter can:
89 1. Download the package from npm (or read local)
90 2. Parse the package.json and extension manifest
91 3. Convert tool definitions to AgentOS ToolDef
92 4. Generate a SkillManifest
93 """
95 format_name = EcosystemFormat.CLAUDE_CODE.value
97 def __init__(self, cache_dir: str | None = None):
98 self._cache_dir = cache_dir or os.path.join(
99 tempfile.gettempdir(), "agentos", "claude_cache"
100 )
101 os.makedirs(self._cache_dir, exist_ok=True)
103 def detect(self, source: str) -> bool:
104 return (
105 source.startswith("claude://")
106 or source.startswith("@") # npm scoped package
107 or "claude-code" in source.lower()
108 or source.endswith(".tgz")
109 )
111 def list_available(self) -> list[str]:
112 """Return known popular Claude Code extensions."""
113 return [
114 "@anthropic/claude-code-tools",
115 "@modelcontextprotocol/server-filesystem",
116 "@modelcontextprotocol/server-github",
117 "@modelcontextprotocol/server-postgres",
118 "@modelcontextprotocol/server-sqlite",
119 "@modelcontextprotocol/server-puppeteer",
120 "@modelcontextprotocol/server-playwright",
121 "@modelcontextprotocol/server-redis",
122 ]
124 def bridge(self, source: str) -> BridgeResult:
125 result = BridgeResult(
126 skill_name=source,
127 source_format=self.format_name,
128 source_uri=source,
129 )
131 try:
132 # Strip protocol prefix
133 if source.startswith("claude://"):
134 source = source[len("claude://") :]
136 # Try to load extension manifest (simulated for now)
137 manifest = self._convert_to_skill(source)
138 if manifest:
139 result.success = True
140 result.manifest = manifest
141 result.skill_name = manifest.name
142 result.warnings.append("Claude Code extension converted to AgentOS format")
143 result.warnings.append(
144 "Note: Some Claude Code extensions use external APIs "
145 "that may require additional configuration"
146 )
147 else:
148 result.error = f"Could not parse Claude Code extension: {source}"
150 except Exception as e:
151 result.error = f"Bridge failed: {e}"
153 return result
155 def _convert_to_skill(self, source: str) -> SkillManifest | None:
156 """Convert a Claude Code extension identifier to a SkillManifest.
158 In production, this would:
159 1. Download the npm package
160 2. Parse package.json for 'claude-code' extension config
161 3. Convert tool definitions
163 For now, we generate a template manifest based on the package name.
164 """
165 name = source.lstrip("@").replace("/", "-").replace("@", "")
166 # Infer tools from package name
167 tools = []
168 if "filesystem" in source.lower():
169 tools.append(
170 ToolDef(
171 name="read_file",
172 description="Read file contents from the filesystem",
173 parameters={
174 "type": "object",
175 "properties": {
176 "path": {"type": "string"},
177 },
178 },
179 )
180 )
181 tools.append(
182 ToolDef(
183 name="write_file",
184 description="Write content to a file",
185 parameters={
186 "type": "object",
187 "properties": {
188 "path": {"type": "string"},
189 "content": {"type": "string"},
190 },
191 },
192 )
193 )
194 elif "github" in source.lower():
195 tools.append(
196 ToolDef(
197 name="github_get_file",
198 description="Get file contents from a GitHub repository",
199 parameters={
200 "type": "object",
201 "properties": {
202 "owner": {"type": "string"},
203 "repo": {"type": "string"},
204 "path": {"type": "string"},
205 },
206 },
207 )
208 )
209 elif "database" in source.lower() or "postgres" in source.lower():
210 tools.append(
211 ToolDef(
212 name="query_database",
213 description="Execute a SQL query against the database",
214 parameters={
215 "type": "object",
216 "properties": {
217 "query": {"type": "string"},
218 },
219 },
220 )
221 )
223 return SkillManifest(
224 name=name,
225 version="1.0.0",
226 description=f"Claude Code extension: {source}",
227 format=SkillFormat.GENERIC,
228 tools=(
229 tools
230 if tools
231 else [
232 ToolDef(
233 name=f"{name}_tool",
234 description=f"Auto-converted tool from {source}",
235 parameters={"type": "object", "properties": {}},
236 )
237 ]
238 ),
239 author="Claude Code Ecosystem",
240 tags=["claude-code", "bridge"],
241 )
244# ── Cursor Adapter ──────────────────────────────────────────────────
247class CursorAdapter(BaseAdapter):
248 """Bridge Cursor rules to AgentOS skills.
250 Cursor uses .cursorrules files and .cursor/rules/ directories
251 to define AI behavior modifications. This adapter converts
252 those rule definitions into AgentOS skills.
253 """
255 format_name = EcosystemFormat.CURSOR.value
257 def detect(self, source: str) -> bool:
258 return (
259 source.startswith("cursor://")
260 or ".cursorrules" in source.lower()
261 or ".cursor/" in source
262 or source.endswith(".mdc")
263 )
265 def list_available(self) -> list[str]:
266 """Return common Cursor rule sources."""
267 return [
268 "cursor://rules/python-best-practices",
269 "cursor://rules/typescript-standards",
270 "cursor://rules/react-patterns",
271 "cursor://rules/testing-guidelines",
272 ]
274 def bridge(self, source: str) -> BridgeResult:
275 result = BridgeResult(
276 skill_name=source,
277 source_format=self.format_name,
278 source_uri=source,
279 )
281 try:
282 if source.startswith("cursor://"):
283 rule_path = source[len("cursor://") :]
284 else:
285 rule_path = source
287 manifest = self._convert_rule(rule_path)
288 if manifest:
289 result.success = True
290 result.manifest = manifest
291 result.skill_name = manifest.name
292 result.warnings.append("Cursor rule converted to AgentOS skill")
293 else:
294 result.error = f"Could not parse Cursor rule: {source}"
296 except Exception as e:
297 result.error = f"Bridge failed: {e}"
299 return result
301 def _convert_rule(self, rule_path: str) -> SkillManifest | None:
302 name = Path(rule_path).stem.replace(".cursorrules", "").replace(".", "-")
303 if not name:
304 name = rule_path.replace("/", "-")
306 return SkillManifest(
307 name=name,
308 version="1.0.0",
309 description=f"Cursor rule: {rule_path}",
310 format=SkillFormat.GENERIC,
311 tools=[],
312 author="Cursor Ecosystem",
313 tags=["cursor", "bridge"],
314 )
317# ── Custom GPT Adapter ──────────────────────────────────────────────
320class CustomGPTAdapter(BaseAdapter):
321 """Bridge Custom GPT instructions to AgentOS skills.
323 Custom GPTs have instructions, conversation starters, knowledge files,
324 and capabilities. This adapter extracts instructions and converts
325 them into an AgentOS skill definition.
326 """
328 format_name = EcosystemFormat.CUSTOM_GPT.value
330 def detect(self, source: str) -> bool:
331 return (
332 source.startswith("gpt://") or "chatgpt.com/g/" in source or source.endswith(".gpt.md")
333 )
335 def list_available(self) -> list[str]:
336 return [
337 "gpt://data-analyst",
338 "gpt://creative-writer",
339 "gpt://code-reviewer",
340 "gpt://research-assistant",
341 ]
343 def bridge(self, source: str) -> BridgeResult:
344 result = BridgeResult(
345 skill_name=source,
346 source_format=self.format_name,
347 source_uri=source,
348 )
350 try:
351 if source.startswith("gpt://"):
352 gpt_id = source[len("gpt://") :]
353 else:
354 gpt_id = source
356 manifest = self._convert_gpt(gpt_id)
357 if manifest:
358 result.success = True
359 result.manifest = manifest
360 result.skill_name = manifest.name
361 result.warnings.append("Custom GPT instructions converted to AgentOS skill")
362 else:
363 result.error = f"Could not parse Custom GPT: {source}"
365 except Exception as e:
366 result.error = f"Bridge failed: {e}"
368 return result
370 def _convert_gpt(self, gpt_id: str) -> SkillManifest | None:
371 name = gpt_id.replace("/", "-").replace(" ", "-")
372 return SkillManifest(
373 name=name,
374 version="1.0.0",
375 description=f"Custom GPT: {gpt_id}",
376 format=SkillFormat.GENERIC,
377 tools=[],
378 author="Custom GPT Ecosystem",
379 tags=["custom-gpt", "bridge"],
380 )
383# ── LangChain Adapter ───────────────────────────────────────────────
386class LangChainAdapter(BaseAdapter):
387 """Bridge LangChain tools to AgentOS skills.
389 LangChain provides a rich ecosystem of tools (toolkits, tools,
390 MCP adapters). This adapter converts them into AgentOS ToolDef
391 and wraps them in a SkillManifest.
392 """
394 format_name = EcosystemFormat.LANGCHAIN.value
396 KNOWN_TOOLS = {
397 "wikipedia": {
398 "name": "wikipedia_query",
399 "description": "Search and retrieve information from Wikipedia",
400 "parameters": {
401 "type": "object",
402 "properties": {
403 "query": {"type": "string", "description": "Search query"},
404 "max_results": {"type": "integer", "default": 3},
405 },
406 "required": ["query"],
407 },
408 },
409 "arxiv": {
410 "name": "arxiv_search",
411 "description": "Search academic papers on arXiv",
412 "parameters": {
413 "type": "object",
414 "properties": {
415 "query": {"type": "string"},
416 "max_results": {"type": "integer", "default": 5},
417 },
418 "required": ["query"],
419 },
420 },
421 "duckduckgo": {
422 "name": "web_search",
423 "description": "Search the web using DuckDuckGo",
424 "parameters": {
425 "type": "object",
426 "properties": {
427 "query": {"type": "string"},
428 },
429 "required": ["query"],
430 },
431 },
432 "python_repl": {
433 "name": "execute_python",
434 "description": "Execute Python code in a REPL environment",
435 "parameters": {
436 "type": "object",
437 "properties": {
438 "code": {"type": "string"},
439 },
440 "required": ["code"],
441 },
442 },
443 "shell": {
444 "name": "execute_shell",
445 "description": "Execute shell commands",
446 "parameters": {
447 "type": "object",
448 "properties": {
449 "command": {"type": "string"},
450 },
451 "required": ["command"],
452 },
453 },
454 }
456 def detect(self, source: str) -> bool:
457 return source.startswith("langchain://") or "langchain" in source.lower()
459 def list_available(self) -> list[str]:
460 return [f"langchain://{name}" for name in self.KNOWN_TOOLS]
462 def bridge(self, source: str) -> BridgeResult:
463 result = BridgeResult(
464 skill_name=source,
465 source_format=self.format_name,
466 source_uri=source,
467 )
469 try:
470 if source.startswith("langchain://"):
471 tool_name = source[len("langchain://") :]
472 else:
473 tool_name = source
475 manifest = self._convert_tool(tool_name)
476 if manifest:
477 result.success = True
478 result.manifest = manifest
479 result.skill_name = manifest.name
480 else:
481 result.error = f"Unknown LangChain tool: {tool_name}"
483 except Exception as e:
484 result.error = f"Bridge failed: {e}"
486 return result
488 def _convert_tool(self, tool_name: str) -> SkillManifest | None:
489 if tool_name not in self.KNOWN_TOOLS:
490 return None
492 tool_info = self.KNOWN_TOOLS[tool_name]
493 tool_def = ToolDef(
494 name=tool_info["name"],
495 description=tool_info["description"],
496 parameters=tool_info["parameters"],
497 )
499 return SkillManifest(
500 name=f"langchain-{tool_name}",
501 version="1.0.0",
502 description=f"LangChain tool: {tool_name}",
503 format=SkillFormat.GENERIC,
504 tools=[tool_def],
505 author="LangChain Ecosystem",
506 tags=["langchain", "bridge"],
507 )
510# ── Adapter Factory ─────────────────────────────────────────────────
513class AdapterFactory:
514 """Factory for creating ecosystem adapters."""
516 _adapters: dict[EcosystemFormat, type] = {}
518 @classmethod
519 def register(cls, fmt: EcosystemFormat, adapter_cls: type):
520 cls._adapters[fmt] = adapter_cls
522 @classmethod
523 def create(cls, fmt: EcosystemFormat, **kwargs) -> BaseAdapter:
524 """Create an adapter for the given ecosystem format."""
525 if fmt not in cls._adapters:
526 raise ValueError(f"Unsupported ecosystem format: {fmt}")
527 return cls._adapters[fmt](**kwargs)
529 @classmethod
530 def detect_format(cls, source: str) -> EcosystemFormat | None:
531 """Auto-detect ecosystem format from source string."""
532 for fmt, adapter_cls in cls._adapters.items():
533 adapter = adapter_cls()
534 if adapter.detect(source):
535 return fmt
536 return None
538 @classmethod
539 def list_supported_formats(cls) -> list[str]:
540 return [f.value for f in cls._adapters]
543# Register built-in adapters
544AdapterFactory.register(EcosystemFormat.CLAUDE_CODE, ClaudeCodeAdapter)
545AdapterFactory.register(EcosystemFormat.CURSOR, CursorAdapter)
546AdapterFactory.register(EcosystemFormat.CUSTOM_GPT, CustomGPTAdapter)
547AdapterFactory.register(EcosystemFormat.LANGCHAIN, LangChainAdapter)
550# ── Ecosystem Bridge (Main Entry) ───────────────────────────────────
553class EcosystemBridge:
554 """Bridge external skill ecosystems into AgentOS SkillRegistry.
556 Usage:
557 bridge = EcosystemBridge()
558 # Single skill
559 result = bridge.bridge("claude://@anthropic/claude-code-tools")
561 # Batch (all available from one ecosystem)
562 results = bridge.bridge_all(EcosystemFormat.CLAUDE_CODE)
564 # Auto-detect and bridge
565 result = bridge.bridge("langchain://wikipedia")
566 """
568 def __init__(self, skill_registry=None):
569 self._skill_registry = skill_registry
570 self._adapters: dict[EcosystemFormat, BaseAdapter] = {}
572 def _get_adapter(self, fmt: EcosystemFormat) -> BaseAdapter:
573 if fmt not in self._adapters:
574 self._adapters[fmt] = AdapterFactory.create(fmt)
575 return self._adapters[fmt]
577 def bridge(self, source: str, fmt: EcosystemFormat | None = None) -> BridgeResult:
578 """Bridge a single skill from external ecosystem.
580 Args:
581 source: Source identifier (e.g., "claude://pkg", "cursor://rule")
582 fmt: Ecosystem format. Auto-detected if not specified.
584 Returns:
585 BridgeResult with converted SkillManifest.
586 """
587 if not fmt:
588 fmt = AdapterFactory.detect_format(source)
589 if not fmt:
590 return BridgeResult(
591 success=False,
592 skill_name=source,
593 error=f"Could not auto-detect ecosystem format for: {source}",
594 )
596 adapter = self._get_adapter(fmt)
597 result = adapter.bridge(source)
599 # Auto-register to skill registry if available
600 if result.success and result.manifest and self._skill_registry:
601 try:
602 self._skill_registry.register_skill(result.manifest)
603 except Exception as e:
604 result.warnings.append(f"Registered but SKillRegistry error: {e}")
606 return result
608 def bridge_all(self, fmt: EcosystemFormat) -> BridgeBatchResult:
609 """Bridge all available skills from an ecosystem."""
610 batch = BridgeBatchResult()
611 adapter = self._get_adapter(fmt)
612 available = adapter.list_available()
614 for source in available:
615 result = adapter.bridge(source)
616 batch.results.append(result)
617 if result.success:
618 batch.succeeded += 1
619 else:
620 batch.failed += 1
621 batch.errors.append(result.error)
623 batch.total = len(available)
624 return batch
626 def batch_bridge(self, sources: list[str]) -> BridgeBatchResult:
627 """Bridge multiple sources, auto-detecting formats."""
628 batch = BridgeBatchResult()
630 for source in sources:
631 result = self.bridge(source)
632 batch.results.append(result)
633 if result.success:
634 batch.succeeded += 1
635 else:
636 batch.failed += 1
637 batch.errors.append(result.error)
639 batch.total = len(sources)
640 return batch
642 def list_available(self, fmt: EcosystemFormat) -> list[str]:
643 """List available skills in an ecosystem."""
644 adapter = self._get_adapter(fmt)
645 return adapter.list_available()
647 def supported_formats(self) -> list[str]:
648 return AdapterFactory.list_supported_formats()