Skip to content

File Sharing

This guide shows you how to share files and data between agents using Artifact Hub.

from acenta import AcentaClient
client = AcentaClient(api_key="...")
# Upload from bytes
with open("report.pdf", "rb") as f:
data = f.read()
artifact = await client.artifact.create(
name="quarterly-report.pdf",
content_type="application/pdf",
data=data,
metadata={
"quarter": "Q4",
"year": 2025,
"department": "engineering"
}
)
print(f"Created: {artifact.id}")
artifact = await client.artifact.upload_file(
path="/path/to/large-file.zip",
name="backup.zip",
metadata={"type": "backup"}
)
artifact = await client.artifact.create_snippet(
name="config.json",
content={
"model": "gpt-4",
"temperature": 0.7,
"max_tokens": 2000
}
)
data = await client.artifact.download(artifact_id="art_xxx")
# Process the data
result = process(data)
await client.artifact.download_file(
artifact_id="art_xxx",
path="/path/to/output.pdf"
)
artifact = await client.artifact.get(artifact_id="art_xxx")
print(f"Name: {artifact.name}")
print(f"Size: {artifact.size_bytes}")
print(f"Type: {artifact.content_type}")
print(f"Created: {artifact.created_at}")

Generate temporary URLs for direct access:

# Get download URL (valid for 1 hour by default)
url = await client.artifact.get_signed_url(
artifact_id="art_xxx",
expires_in=3600
)
# Use in external systems
print(f"Download URL: {url}")

For large files, get a signed upload URL:

upload = await client.artifact.get_upload_url(
name="large-file.bin",
content_type="application/octet-stream",
size_bytes=1073741824 # 1 GB
)
# Upload directly to the URL
import httpx
async with httpx.AsyncClient() as http:
with open("large-file.bin", "rb") as f:
await http.put(upload.url, content=f)
# Confirm upload
artifact = await client.artifact.confirm_upload(upload.id)
# Original version
artifact = await client.artifact.create(
name="model.pt",
data=model_v1_bytes
)
# Update creates a new version
artifact = await client.artifact.update(
artifact_id=artifact.id,
data=model_v2_bytes,
metadata={"version": "2.0"}
)
versions = await client.artifact.list_versions(artifact_id=artifact.id)
for v in versions:
print(f"Version {v.version}")
print(f" Created: {v.created_at}")
print(f" Size: {v.size_bytes}")
data = await client.artifact.download(
artifact_id=artifact.id,
version=1 # Get version 1
)
await client.artifact.share(
artifact_id="art_xxx",
with_agents=["agent-a", "agent-b", "agent-c"]
)
await client.artifact.share(
artifact_id="art_xxx",
with_groups=["data-science-team", "ml-team"]
)
access = await client.artifact.get_access(artifact_id="art_xxx")
print(f"Shared with agents: {access.agents}")
print(f"Shared with groups: {access.groups}")
await client.artifact.unshare(
artifact_id="art_xxx",
agents=["agent-a"],
groups=["data-science-team"]
)
artifacts = await client.artifact.list()
for a in artifacts:
print(f"{a.id}: {a.name} ({a.size_bytes} bytes)")
# Filter by metadata
artifacts = await client.artifact.list(
metadata_filter={"department": "engineering"}
)
# Filter by content type
artifacts = await client.artifact.list(
content_type="application/pdf"
)
# Pagination
artifacts = await client.artifact.list(
limit=10,
offset=20
)
shared = await client.artifact.list_shared()
for a in shared:
print(f"{a.name} (from {a.owner})")
await client.artifact.delete(artifact_id="art_xxx")
await client.artifact.delete_version(
artifact_id="art_xxx",
version=1
)

Use artifacts in Coordination Hub workflows:

{
"id": "download_data",
"type": "agent",
"agent": "data-loader",
"input": {
"artifact_id": "{{ input.artifact_id }}"
}
},
{
"id": "process",
"type": "agent",
"agent": "processor",
"depends_on": ["download_data"],
"input": {
"data": "{{ steps.download_data.output.data }}"
}
},
{
"id": "save_result",
"type": "agent",
"agent": "artifact-saver",
"depends_on": ["process"],
"input": {
"result": "{{ steps.process.output }}",
"name": "processed-{{ input.artifact_id }}.json"
}
}